使用變更摘要估算器

發行項
08/15/2024

適用於：NoSQL

本文說明如何在變更摘要處理器執行個體讀取變更摘要時，監視處理器的進度。

為什麼監視進度很重要？

變更摘要處理器會以指標形式在變更摘要中向前移動，並將變更傳遞至委派執行。

您的變更摘要處理器部署可以根據其可用資源 (例如 CPU、記憶體、網路等等)，以特定的速率處理變更。

如果此速率比您在 Azure Cosmos DB 容器中發生變更的速率慢，您的處理器就會開始落後。

識別此情況有助於了解我們是否需要調整變更摘要處理器的部署。

實作變更摘要估算器

.NET
Java

做為自動通知的推送模型

如同變更摘要處理器，變更摘要估算器可做為推送模型來使用。估算器會測量上次處理的項目 (由租用容器的狀態所定義) 與容器中最新變更之間的差異，並將此值推送至委派。測量的執行間隔也可以自訂，預設值為 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 分鐘之間的頻率是不錯的起點，頻率越低，所使用的要求單位耗用就越高。

此範例會使用最新版本模式中的變更摘要處理器，並估計處理變更摘要事件的延遲。在應用程式中：文件正在插入至一個容器 (「摘要容器」)，同時另一個背景工作執行緒或背景工作應用程式正在從摘要容器的變更摘要中提取所插入的文件，並在其上以某種方式操作。

變更摘要處理器如下進行建置和啟動：

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()");
    };
}

支援的變更摘要模式

變更摘要估算器可用於最新版本模式，以及所有版本和刪除模式。在這兩種模式中，提供的估計不保證是待處理的未完成變更的確切數量。

其他資源

下一步

您現在可以在下列文章中繼續深入了解變更摘要處理器：

變更摘要處理器的概觀

共用方式為

使用變更摘要估算器

為什麼監視進度很重要？

實作變更摘要估算器

做為自動通知的推送模型

做為隨選詳細估算

估算器部署

支援的變更摘要模式

其他資源

下一步

意見反應

其他資源