変更フィード推定ツールを使用する

自動通知のプッシュ モデルとして

変更フィード プロセッサと同様に、変更フィード推定機能はプッシュ モデルとして機能します。 エスティメーターは、最後に処理された項目 (リース コンテナーの状態によって定義されます) とコンテナーの最新の変更の差を測定し、この値をデリゲートにプッシュします。 測定の間隔は、既定値の 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 と同じ名前を共有する場合。

他の 2 つのパラメーターはデリゲートであり、プロセッサによって 読み取られる保留中の変更の数 と、この測定を実行する時間間隔を表す数値を受け取ります。

推定を受け取るデリゲートの例を次に示します。

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 つのエスティメーター インスタンスは、変更フィード プロセッサのデプロイ内のすべてのリースとインスタンスの進行状況を追跡できます。

各見積もりでは、監視対象コンテナーとリース コンテナーからの要求ユニットが消費されます。 1分間隔の頻度は良い開始点です。頻度が低いほど、要求ユニットの消費量が増えます。

この例では、変更フィード プロセッサを最新バージョン モードで使用し、変更フィード イベントの処理におけるラグの推定を行います。 アプリケーションでは、ドキュメントが 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());

最初の合計ラグは 0 で、最後に作成されたドキュメントの数以上にする必要があります。 合計ラグ値はログに記録することも、さらに分析するために使用することもできます。これにより、変更フィード処理のパフォーマンスを監視し、システムの潜在的なボトルネックや遅延を特定できます。

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