Обработчик потока изменений в Azure Cosmos DB

Обработчик канала изменений в .NET доступен для режима последней версии и режима всех версий и удалений. Все версии и режим удаления доступны в режиме предварительного просмотра и поддерживаются для обработчика канала изменений, начиная с версии 3.40.0-preview.0. Точка входа для обоих режимов всегда является отслеживаемой контейнерой.

Чтобы читать в режиме последней версии в экземпляре Container, вызовите GetChangeFeedProcessorBuilder:

/// <summary>
/// Start the Change Feed Processor to listen for changes and process them with the HandleChangesAsync implementation.
/// </summary>
private static async Task<ChangeFeedProcessor> StartChangeFeedProcessorAsync(
    CosmosClient cosmosClient,
    IConfiguration configuration)
{
    string databaseName = configuration["SourceDatabaseName"];
    string sourceContainerName = configuration["SourceContainerName"];
    string leaseContainerName = configuration["LeasesContainerName"];

    Container leaseContainer = cosmosClient.GetContainer(databaseName, leaseContainerName);
    ChangeFeedProcessor changeFeedProcessor = cosmosClient.GetContainer(databaseName, sourceContainerName)
        .GetChangeFeedProcessorBuilder<ToDoItem>(processorName: "changeFeedSample", onChangesDelegate: HandleChangesAsync)
            .WithInstanceName("consoleHost")
            .WithLeaseContainer(leaseContainer)
            .Build();

    Console.WriteLine("Starting Change Feed Processor...");
    await changeFeedProcessor.StartAsync();
    Console.WriteLine("Change Feed Processor started.");
    return changeFeedProcessor;
}

Чтобы читать, используя все версии и режим удалений, вызовите GetChangeFeedProcessorBuilderWithAllVersionsAndDeletes из экземпляра Container:

Container leaseContainer = client.GetContainer(Program.databaseName, Program.leasesContainer);
Container monitoredContainer = client.GetContainer(Program.databaseName, containerName);
ChangeFeedProcessor changeFeedProcessor = monitoredContainer
    .GetChangeFeedProcessorBuilderWithAllVersionsAndDeletes<ToDoItem>(processorName: "changeFeedBasic", onChangesDelegate: Program.HandleChangesAsync)
        .WithInstanceName("consoleHost")
        .WithLeaseContainer(leaseContainer)
        .Build();

Для обоих режимов первый параметр — это отдельное имя, описывающее цель этого процессора. Вторая часть имени — это реализация делегата, которая обрабатывает изменения данных.

Вот пример делегата для режима последней версии:

/// <summary>
/// The delegate receives batches of changes as they are generated in the change feed and can process them.
/// </summary>
static async Task HandleChangesAsync(
    ChangeFeedProcessorContext context,
    IReadOnlyCollection<ToDoItem> changes,
    CancellationToken cancellationToken)
{
    Console.WriteLine($"Started handling changes for lease {context.LeaseToken}...");
    Console.WriteLine($"Change Feed request consumed {context.Headers.RequestCharge} RU.");
    // SessionToken if needed to enforce Session consistency on another client instance
    Console.WriteLine($"SessionToken ${context.Headers.Session}");

    // We may want to track any operation's Diagnostics that took longer than some threshold
    if (context.Diagnostics.GetClientElapsedTime() > TimeSpan.FromSeconds(1))
    {
        Console.WriteLine($"Change Feed request took longer than expected. Diagnostics:" + context.Diagnostics.ToString());
    }

    foreach (ToDoItem item in changes)
    {
        Console.WriteLine($"Detected operation for item with id {item.id}, created at {item.creationTime}.");
        // Simulate some asynchronous operation
        await Task.Delay(10);
    }

    Console.WriteLine("Finished handling changes.");
}

Вот пример делегата для всех версий и режима удаления:

static async Task HandleChangesAsync(ChangeFeedProcessorContext context, IReadOnlyCollection<ChangeFeedItem<ToDoItem>> changes, CancellationToken cancellationToken)
{
    Console.WriteLine($"Started handling changes for lease {context.LeaseToken}...");
    Console.WriteLine($"Change Feed request consumed {context.Headers.RequestCharge} RU.");
    // SessionToken if needed to enforce Session consistency on another client instance
    Console.WriteLine($"SessionToken ${context.Headers.Session}");

    // We may want to track any operation's Diagnostics that took longer than some threshold
    if (context.Diagnostics.GetClientElapsedTime() > TimeSpan.FromSeconds(1))
    {
        Console.WriteLine($"Change Feed request took longer than expected. Diagnostics:" + context.Diagnostics.ToString());
    }

    foreach (ChangeFeedItem<ToDoItem> item in changes)
    {
        if (item.Metadata.OperationType == ChangeFeedOperationType.Delete)
        {
            Console.WriteLine($"\tDetected {item.Metadata.OperationType} operation for item.");
        }
        else
        {
            Console.WriteLine($"\tDetected {item.Metadata.OperationType} operation for item with id {item.Current.id}.");
        }
        // Simulate work
        await Task.Delay(1);
    }
}

После этого вы определите имя вычислительного экземпляра или уникальный идентификатор с помощью WithInstanceName. Имя вычислительного экземпляра должно быть уникальным и отличаться для каждого экземпляра, который вы разворачиваете. Контейнер задается для поддержания состояния аренды с помощью WithLeaseContainer.

Вызов Build предоставляет экземпляр процессора, который можно запустить с помощью вызова StartAsync.

Жизненный цикл обработки

Обычный жизненный цикл хост-инстанса:

1. Прочитайте канал изменений.
2. Если изменений нет, приостановите выполнение на предопределенное время (настраиваемое с помощью WithPollInterval в Builder) и вернитесь к пункту №1.
3. При наличии изменений отправьте их ответственному лицу.
4. Когда делегат успешно завершит обработку изменений, обновите хранилище данных аренды с последней обработанной точкой во времени и перейдите к пункту № 1.

Обработка ошибок

Обработчик канала изменений устойчив к ошибкам пользовательского кода. Если реализация делегата имеет необработанное исключение (шаг 4), поток, обрабатывающий конкретный пакет изменений, останавливается, а новый поток в конечном итоге создается. Новый поток проверяет последнее сохраненное время в хранилище орендных записей для данного диапазона значений ключа раздела. Новый поток перезапускается оттуда, эффективно отправляя тот же пакет изменений делегату. Это поведение продолжается до тех пор, пока делегат не обрабатывает изменения правильно, и это причина, по которой обработчик канала изменений имеет гарантию "по крайней мере один раз".

Чтобы предотвратить непрерывное повторение одного и того же пакета изменений обработчиком канала изменений, необходимо добавить логику в код делегата для записи документов, за исключением исключения, в очередь сообщений с ошибками. Такая схема гарантирует, что вы сможете отследить необработанные изменения, сохранив возможность продолжать обработку будущих изменений. Очередь сообщений об ошибках может быть другим контейнером Azure Cosmos DB. Точное хранилище данных не имеет значения. Вы просто хотите сохранить необработанные изменения.

Вы также можете использовать оценщик потока изменений для отслеживания хода выполнения экземпляров обработчика потока изменений при чтении потока изменений, или использовать уведомления о жизненном цикле для обнаружения скрытых сбоев.

Уведомления жизненного цикла

Обработчик канала изменений можно подключить к любому соответствующему событию в жизненном цикле. Вы можете получать уведомления одному или всем из них. Мы рекомендуем зарегистрировать по меньшей мере получение уведомлений об ошибках.

• Зарегистрируйте обработчик для WithLeaseAcquireNotification, чтобы получать уведомления о том, что текущий узел зарегистрировал аренду и начал ее обрабатывать.
• Зарегистрируйте обработчик для WithLeaseReleaseNotification, чтобы получать уведомления, когда текущий сервер освобождает аренду и прекращает её обработку.
• Зарегистрируйте обработчик для WithErrorNotification, чтобы получать уведомления, когда текущий хост обнаруживает исключение во время обработки. Необходимо иметь возможность различать, является ли источник делегатом пользователя (необработанным исключением) или ошибкой, с которой обработчик сталкивается при попытке получить доступ к отслеживаемой контейнере (например, сетевым проблемам).

Уведомления жизненного цикла доступны в обоих режимах канала изменений. Ниже приведен пример уведомлений жизненного цикла в режиме последней версии:

Container.ChangeFeedMonitorLeaseAcquireDelegate onLeaseAcquiredAsync = (string leaseToken) =>
{
    Console.WriteLine($"Lease {leaseToken} is acquired and will start processing");
    return Task.CompletedTask;
};

Container.ChangeFeedMonitorLeaseReleaseDelegate onLeaseReleaseAsync = (string leaseToken) =>
{
    Console.WriteLine($"Lease {leaseToken} is released and processing is stopped");
    return Task.CompletedTask;
};

Container.ChangeFeedMonitorErrorDelegate onErrorAsync = (string LeaseToken, Exception exception) =>
{
    if (exception is ChangeFeedProcessorUserException userException)
    {
        Console.WriteLine($"Lease {LeaseToken} processing failed with unhandled exception from user delegate {userException.InnerException}");
    }
    else
    {
        Console.WriteLine($"Lease {LeaseToken} failed with {exception}");
    }

    return Task.CompletedTask;
};

ChangeFeedProcessor changeFeedProcessor = monitoredContainer
    .GetChangeFeedProcessorBuilder<ToDoItem>("changeFeedNotifications", handleChanges)
        .WithLeaseAcquireNotification(onLeaseAcquiredAsync)
        .WithLeaseReleaseNotification(onLeaseReleaseAsync)
        .WithErrorNotification(onErrorAsync)
        .WithInstanceName("consoleHost")
        .WithLeaseContainer(leaseContainer)
        .Build();

Единица развертывания

Одна единица развертывания обработчика канала изменений состоит из одного или нескольких вычислительных экземпляров, имеющих одинаковое значение processorName и одну конфигурацию контейнера аренды, но разные имена экземпляров. Вы можете иметь множество единиц развертывания, в которых каждый блок имеет другой бизнес-поток для изменений, и каждая единица развертывания состоит из одного или нескольких экземпляров.

Например, у вас может быть одна единица развертывания, которая активирует внешний API при каждом изменении контейнера. Другая единица развертывания может перемещать данные в режиме реального времени при каждом изменении. Когда изменение происходит в отслеживаемом контейнере, все единицы развертывания уведомляются.

Динамическое масштабирование

Как упоминалось ранее, в единице развертывания можно использовать один или несколько вычислительных экземпляров. Чтобы воспользоваться преимуществами распределения вычислений в единице развертывания, единственными ключевыми требованиями являются следующие:

• Все экземпляры должны иметь одинаковую конфигурацию арендуемых контейнеров.
• У всех экземпляров должно быть одинаковое значение processorName.
• Каждый экземпляр должен иметь уникальное имя (WithInstanceName).

Если применяются эти три условия, обработчик канала изменений распределяет все аренды, которые находятся в контейнере аренды во всех запущенных экземплярах этой единицы развертывания, и он параллелизирует вычисления с помощью алгоритма равного распределения. Аренда принадлежит одному экземпляру в любое время, поэтому количество экземпляров не должно превышать количество аренд.

Число экземпляров может увеличиваться и уменьшаться. Обработчик потока изменений динамически настраивает нагрузку, соответственно перераспределяя её.

Кроме того, обработчик канала изменений может динамически корректировать масштаб контейнера при увеличении его пропускной способности или объёма хранилища. Когда контейнер растет, процессор изменений прозрачно обрабатывает сценарий путем автоматического увеличения количества аренд и распределения новых аренды между существующими экземплярами.

Время запуска

По умолчанию, при первом запуске обработчика канала изменений, он инициализирует контейнер аренды и запускает его жизненный цикл обработки. Изменения, которые произошли в отслеживаемом контейнере перед первой инициализацией обработчика канала изменений, не будут обнаружены.

Чтение с предыдущей даты и времени

Обработчик изменений можно инициализировать для чтения изменений, начиная с определенной даты и времени, передав экземпляр DateTime в расширение построителя WithStartTime:

Container leaseContainer = client.GetContainer(databaseId, Program.leasesContainer);
Container monitoredContainer = client.GetContainer(databaseId, Program.monitoredContainer);
ChangeFeedProcessor changeFeedProcessor = monitoredContainer
    .GetChangeFeedProcessorBuilder<ToDoItem>("changeFeedTime", Program.HandleChangesAsync)
        .WithInstanceName("consoleHost")
        .WithLeaseContainer(leaseContainer)
        .WithStartTime(particularPointInTime)
        .Build();

Обработчик потока изменений инициализируется для указанной даты и времени и начинает считывать изменения, которые произошли после этого.

Чтение с начала

В других сценариях, например при миграции данных или при анализе всей истории контейнера, необходимо считывать канал изменений с начала существования этого контейнера. Вы можете использовать WithStartTime расширение сборщика, но передать DateTime.MinValue.ToUniversalTime(), которое создаёт представление минимального значения DateTime в формате UTC; как в следующем примере:

Container leaseContainer = client.GetContainer(databaseId, Program.leasesContainer);
Container monitoredContainer = client.GetContainer(databaseId, Program.monitoredContainer);
ChangeFeedProcessor changeFeedProcessor = monitoredContainer
    .GetChangeFeedProcessorBuilder<ToDoItem>("changeFeedBeginning", Program.HandleChangesAsync)
        .WithInstanceName("consoleHost")
        .WithLeaseContainer(leaseContainer)
        .WithStartTime(DateTime.MinValue.ToUniversalTime())
        .Build();

Обработчик канала изменений инициализирован и начинает считывать изменения с начала времени существования контейнера.

Полные рабочие примеры см . здесь. Пример реализации делегата при чтении канала изменений в режиме последней версии:

private static Consumer<List<JsonNode>> handleChanges() {
    return (List<JsonNode> docs) -> {
        logger.info("Start handleChanges()");

        for (JsonNode document : docs) {
            try {
                //Change Feed hands the document to you in the form of a JsonNode
                //As a developer you have two options for handling the JsonNode document provided to you by Change Feed
                //One option is to operate on the document in the form of a JsonNode, as shown below. This is great
                //especially if you do not have a single uniform data model for all documents.
                logger.info("Document received: " + OBJECT_MAPPER.writerWithDefaultPrettyPrinter()
                        .writeValueAsString(document));

                //You can also transform the JsonNode to a POJO having the same structure as the JsonNode,
                //as shown below. Then you can operate on the POJO.
                CustomPOJO2 pojo_doc = OBJECT_MAPPER.treeToValue(document, CustomPOJO2.class);
                logger.info("id: " + pojo_doc.getId());

            } catch (JsonProcessingException e) {
                e.printStackTrace();
            }
        }
        isWorkCompleted = true;
        logger.info("End handleChanges()");

    };
}

options = new ChangeFeedProcessorOptions();
options.setStartFromBeginning(false);
options.setLeasePrefix("myChangeFeedDeploymentUnit");
options.setFeedPollDelay(Duration.ofSeconds(5));
options.setFeedPollThroughputControlConfig(throughputControlGroupConfig);

Реализация делегата для чтения веб-канала изменений во всех версиях и удалении аналогична, но вместо вызова .handleChanges(), вызовите .handleAllVersionsAndDeletesChanges(). Все версии и режим удаления находятся в предварительной версии и доступны в версии Java SDK >= 4.42.0.

Приведем пример:

public static ChangeFeedProcessor getChangeFeedProcessorForAllVersionsAndDeletesMode(String hostName, CosmosAsyncContainer feedContainer, CosmosAsyncContainer leaseContainer) {
    return new ChangeFeedProcessorBuilder()
            .hostName(hostName)
            .options(options)
            .feedContainer(feedContainer)
            .leaseContainer(leaseContainer)
            .handleAllVersionsAndDeletesChanges((List<ChangeFeedProcessorItem> changeFeedProcessorItems) -> {
                logger.info("--->handleAllVersionsAndDeletesChanges() START");

                for (ChangeFeedProcessorItem item : changeFeedProcessorItems) {
                    try {
                        // AllVersionsAndDeletes Change Feed hands the document to you in the form of ChangeFeedProcessorItem
                        // As a developer you have two options for handling the ChangeFeedProcessorItem provided to you by Change Feed
                        // One option is to operate on the item as it is and call the different getters for different states, as shown below.
                        logger.info("---->DOCUMENT RECEIVED: {}", OBJECT_MAPPER.writerWithDefaultPrettyPrinter()
                                .writeValueAsString(item));

                        logger.info("---->CURRENT RECEIVED: {}", item.getCurrent());
                        logger.info("---->PREVIOUS RECEIVED: {}", item.getPrevious());
                        logger.info("---->METADATA RECEIVED: {}", item.getChangeFeedMetaData());

                        // You can also transform the ChangeFeedProcessorItem to JsonNode and work on the generic json structure.
                        // This is great especially if you do not have a single uniform data model for all documents.
                        logger.info("----=>JsonNode received: " + item.toJsonNode());

                    } catch (JsonProcessingException e) {
                        e.printStackTrace();
                    }
                }
                logger.info("--->handleAllVersionsAndDeletesChanges() END");

            })
            .buildChangeFeedProcessor();
}

В любом режиме канала изменений его можно присвоить changeFeedProcessorInstance и передать параметры имени вычислительного экземпляра hostName, отслеживаемого контейнера, называемого здесь feedContainer, и leaseContainer. Затем запустите обработчик канала изменений:

logger.info("Start Change Feed Processor on worker (handles changes asynchronously)");
ChangeFeedProcessor changeFeedProcessorInstance = new ChangeFeedProcessorBuilder()
    .hostName("SampleHost_1")
    .feedContainer(feedContainer)
    .leaseContainer(leaseContainer)
    .handleChanges(handleChanges())
    .options(options)
    .buildChangeFeedProcessor();
changeFeedProcessorInstance.start()
                           .subscribeOn(Schedulers.boundedElastic())
                           .subscribe();

Жизненный цикл обработки

Обычный жизненный цикл экземпляра хоста:

1. Прочтите канал изменений.
2. Если изменений нет, подождите заданное время (которое можно настроить с помощью options.setFeedPollDelay в конструкторе) и вернитесь к #1.
3. Если есть изменения, отправьте их уполномоченному.
4. После успешной обработки изменений делегат обновите хранилище аренды с помощью последней обработанной точки во времени и перейдите в #1.

Обработка ошибок

Обработчик потока изменений устойчив к ошибкам пользовательского кода. Если реализация делегата имеет необработанное исключение (шаг 4), поток, обрабатывающий конкретный пакет изменений, останавливается, и создается новый поток. Новый поток проверяет последнюю точку во времени, которую хранилище арендных данных сохранило для этого диапазона значений ключа раздела, и он перезапускается с этого момента, эффективно отправляя тот же пакет изменений делегату. Это поведение продолжается до тех пор, пока делегат не обрабатывает изменения правильно. Именно поэтому обработчик ленты изменений гарантирует выполнение "как минимум один раз".

Чтобы предотвратить «застревание» обработчика потока изменений, который непрерывно повторяет один и тот же пакет изменений, необходимо добавить логику в код делегата для записи документов в ошибочное сообщение в случае возникновения исключения. Такая схема гарантирует, что вы сможете отследить необработанные изменения, сохранив возможность продолжать обработку будущих изменений. Сообщение об ошибке может быть другим контейнером Azure Cosmos DB. Точное хранилище данных не имеет значения. Вы просто хотите сохранить необработанные изменения.

Вы также можете использовать инструмент оценки потока изменений для отслеживания хода выполнения экземпляров обработчика потока изменений при их чтении.

Единица развертывания

Одна единица развертывания обработчика канала изменений состоит из одного или нескольких вычислительных экземпляров с одинаковой конфигурацией контейнера аренды leasePrefix, но с разными значениями hostName. Вы можете иметь множество единиц развертывания, в которых каждый из них имеет другой бизнес-поток для изменений, и каждая единица развертывания состоит из одного или нескольких экземпляров.

Например, у вас может быть одна единица развертывания, которая активирует внешний API при каждом изменении контейнера. Другая единица развертывания может перемещать данные в режиме реального времени при каждом изменении. Когда изменение происходит в отслеживаемом контейнере, все единицы развертывания уведомляются.

Динамическое масштабирование

Как упоминалось ранее, в единице развертывания можно использовать один или несколько вычислительных экземпляров. Чтобы воспользоваться преимуществами распределения вычислений в единице развертывания, единственными ключевыми требованиями являются следующие:

• Все экземпляры должны иметь одинаковую конфигурацию арендного контейнера.
• Все экземпляры должны иметь одинаковое значение в options.setLeasePrefix (или вообще не иметь заданного значения).
• У каждого экземпляра должен быть разный hostName.

Если применяются эти три условия, обработчик канала изменений распределяет все арендные соглашения в контейнере арендных соглашений между всеми работающими экземплярами этой единицы развертывания, параллелизируя вычисления с использованием алгоритма равномерного распределения. Аренда принадлежит одному экземпляру в любое время, поэтому количество экземпляров не должно превышать количество аренд.

Число экземпляров может увеличиваться и уменьшаться. Обработчик изменяющегося потока динамически настраивает нагрузку, перераспределяя ее соответствующим образом. Единицы развертывания могут совместно использовать тот же самый контейнер аренды, но каждая из них должна иметь другое leasePrefix значение.

Кроме того, обработчик потоков изменений может динамически масштабировать контейнер, если увеличивается его пропускная способность или объем хранилища. Когда контейнер растет, обработчик канала изменений прозрачно обрабатывает сценарий путем динамического увеличения аренды и распределения новых аренд между существующими экземплярами.

Время запуска

По умолчанию при первом запуске обработчика канала изменений он инициализирует контейнер арендных записей и запускает его жизненный цикл обработки. Любые изменения, произошедшие в отслеживаемом контейнере до инициализации обработчика канала изменений в первый раз, не обнаружены.

Чтение с предыдущей даты и времени

Обработчик канала изменений можно инициализировать для чтения изменений, начиная с определенной даты и времени, установив setStartTime в options. Обработчик канала изменений инициализируется для указанной даты и времени, и начинает считывать изменения, которые произошли после этого.

Чтение с начала

В примере setStartFromBeginning задано значение false, которое совпадает со значением по умолчанию. В других сценариях, например при миграции данных или при анализе всей истории контейнера, необходимо считывать канал изменений с начала существования этого контейнера. Для этого можно задать значение setStartFromBeginningtrue. Обработчик канала изменений инициализирован и начинает считывать изменения с начала времени существования контейнера.