使用變更饋送估算器

作為自動通知的即時推送模型

與 變更饋源處理器類似，變更饋源估計器可作為推送模型運作。 估計器會衡量最後處理的項目（由租約容器狀態定義）與容器最新變更之間的差值，並將此值推送給代理。 測量的間隔也可以自訂，預設值為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 及名稱。

另外兩個參數是委派(delegation)，其接收代表處理器待讀取的變更數量的數值，以及您希望進行測量的時間間隔。

一個接收估計值的代表範例如下：

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