你当前正在访问 Microsoft Azure Global Edition 技术文档网站。 如果需要访问由世纪互联运营的 Microsoft Azure 中国技术文档网站，请访问 https://docs.azure.cn。

使用更改源估算器

适用范围： NoSQL

本文介绍如何监视更改源处理器实例在读取更改源时的进度。

为什么监视进度很重要？

更改源处理器充当一个指针，它会在更改源中向前移动，并将更改传递给委托实现。

更改源处理器部署可以按特定速率处理更改，具体取决于可用的资源，例如 CPU、内存、网络，等等。

如果该速率低于更改在 Azure Cosmos DB 容器中的发生速率，则处理器就会开始出现滞后现象。

确定此情况后，我们就可以决定是否需要缩放更改源处理器部署。

实现更改源估算器

.NET

作为自动通知的推送模型

与更改源处理器一样，更改源估算器可用作推送模型。 此估算器会度量上一个处理项（按租用容器的状态定义）和容器中的最新更改之间的差异，并将该值推送到某个委托。 进行度量的时间间隔也可以自定义，默认值为 5 秒。

例如，如果更改源处理器定义如下：

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

那么，若要初始化某个估算器来度量该处理器，正确的方式是使用 GetChangeFeedEstimatorBuilder，如下所示：

ChangeFeedProcessor changeFeedEstimator = monitoredContainer
    .GetChangeFeedEstimatorBuilder("changeFeedEstimator", Program.HandleEstimationAsync, TimeSpan.FromMilliseconds(1000))
    .WithLeaseContainer(leaseContainer)
    .Build();

其中的处理器和估算器共享同一 leaseContainer 和同一名称。

另外有两个参数，一个是委托，其接收的数字表示有多少更改待处理器读取，另一个是需要进行该度量的时间间隔。

例如，下面是用于接收估算的委托：

static async Task HandleEstimationAsync(long estimation, CancellationToken cancellationToken)
{
    if (estimation > 0)
    {
        Console.WriteLine($"\tEstimator detected {estimation} items pending to be read by the Processor.");
    }

    await Task.Delay(0);
}

可以将此估算发送给监视解决方案，并通过它了解进度随时间的变化情况。

作为按需详细估算

有一种替代方法可让你按需获得估算值（与推送模型相比）。 此模型还提供了更多详细信息：

• 每次租用的估计滞后时间。
• 实例拥有并处理每次租用，因此你可以确定该实例是否存在问题。

如果更改源处理器定义如下：

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

你可以使用相同的租用配置创建估算器：

ChangeFeedEstimator changeFeedEstimator = monitoredContainer
    .GetChangeFeedEstimator("changeFeedEstimator", leaseContainer);

无论何时，只要你需要，就可以以所需的频率获得详细的估算值：

Console.WriteLine("Checking estimation...");
using FeedIterator<ChangeFeedProcessorState> estimatorIterator = changeFeedEstimator.GetCurrentStateIterator();
while (estimatorIterator.HasMoreResults)
{
    FeedResponse<ChangeFeedProcessorState> states = await estimatorIterator.ReadNextAsync();
    foreach (ChangeFeedProcessorState leaseState in states)
    {
        string host = leaseState.InstanceName == null ? $"not owned by any host currently" : $"owned by host {leaseState.InstanceName}";
        Console.WriteLine($"Lease [{leaseState.LeaseToken}] {host} reports {leaseState.EstimatedLag} as estimated lag.");
    }
}

每个 ChangeFeedProcessorState 都会包含租用和滞后时间信息，以及当前哪个实例拥有它。

估算器部署

更改源估算器不需部署在更改源处理器中，也不需部署在同一项目中。 建议在与处理器完全不同的独立实例上部署估算器。 单个估算器实例可以跟踪更改源处理器部署中所有租约和实例的进度。

每次估算都将消耗受监视容器和租用容器的请求单位。 间隔 1 分钟的频率比较合适，频率越低，消耗的请求单位数就越高。

Java

提供的示例表示一个示例 Java 应用程序，该应用程序演示更改源处理器的实现，并估算了在处理更改源事件过程中的滞后时间。 在应用程序中，文档被插入到一个容器（“源容器”），同时，另一个工作线程或辅助角色应用程序正在从源容器的更改源中拉取插入的文档，并以某种方式处理这些文档。

更改源处理器的生成和启动如下所示：

ChangeFeedProcessor changeFeedProcessorMainInstance = new ChangeFeedProcessorBuilder()
    .hostName("SampleHost_1")
    .feedContainer(feedContainer)
    .leaseContainer(leaseContainer)
    .handleChanges(handleChangesWithLag())
    .buildChangeFeedProcessor();

try {
    changeFeedProcessorMainInstance
        .start()
        .subscribeOn(Schedulers.boundedElastic())
        .timeout(Duration.ofSeconds(10))
        .then(Mono.just(changeFeedProcessorMainInstance)
                  .delayElement(Duration.ofSeconds(10))
                  .flatMap(value -> changeFeedProcessorMainInstance
                      .stop()
                      .subscribeOn(Schedulers.boundedElastic())
                      .timeout(Duration.ofSeconds(10))))
        .subscribe();
} catch (Exception ex) {
    logger.error("Change feed processor did not start and stopped in the expected time", ex);
    throw ex;
}

只要使用相同的输入容器（feedContainer 和 leaseContainer）和完全相同的租约前缀 (CONTAINER_NAME + "-lease")，就可以在单独的应用程序（例如运行状况监视器）上执行更改源处理器滞后检查。 估算器代码要求更改源处理器有机会完全初始化 leaseContainer 的文档。

估算器通过检索更改源处理器的当前状态并将正在处理的每个事件的估算滞后时间值求和来计算累积滞后时间：

Mono<List<ChangeFeedProcessorState>> currentState = changeFeedProcessorMainInstance.getCurrentState();
currentState.map(state -> {
    for (ChangeFeedProcessorState changeFeedProcessorState : state) {
        totalLag.addAndGet(changeFeedProcessorState.getEstimatedLag());
    }
    return state;
}).subscribe();

// Initially totalLag should be zero
logger.info("Initially total lag is : {}", totalLag.get());

totalLag.set(0);
currentState = changeFeedProcessorSideCart.getCurrentState();
currentState.map(state -> {
    for (ChangeFeedProcessorState changeFeedProcessorState : state) {
        totalLag.addAndGet(changeFeedProcessorState.getEstimatedLag());
    }
    return state;
}).subscribe();

// Initially totalLag should be zero
logger.info("Initially total lag is : {}", totalLag.get());

总滞后时间最初应为零，最后应大于或等于创建的文档数。 可以记录或使用总滞后时间值作进一步分析，从而可以监视更改源处理的效果，找出系统中任何潜在的瓶颈或滞后情况：

totalLag.set(0);
changeFeedProcessorSideCart.start().block();
currentState = changeFeedProcessorSideCart.getCurrentState();
currentState.map(state -> {
    for (ChangeFeedProcessorState changeFeedProcessorState : state) {
        totalLag.addAndGet(changeFeedProcessorState.getEstimatedLag());
    }
    return state;
}).block();

// Finally, totalLag should be greater or equal to the number of documents created
logger.info("Finally total lag is : {}", totalLag.get());

接收更改并滞后处理更改的委托的示例如下：

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

        try {
            Thread.sleep(Duration.ofSeconds(5).toMillis());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        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();
            }
        }
        logger.info("End handleChangesWithLag()");
    };
}

其他资源

• Azure Cosmos DB SDK
• GitHub 上的用法示例 (.NET)
• GitHub 上的用法示例 (Java)
• GitHub 上的其他示例

后续步骤

现在，可以通过以下文章继续详细了解更改源处理器：

• 更改源处理器概述