Verwenden Sie den Änderungsfeed-Schätzer

Als Pushmodell für automatische Benachrichtigungen

Wie der Änderungsfeedprozessor kann der Änderungsfeed-Schätzer als Pushmodell funktionieren. Der Schätzer misst den Unterschied zwischen dem zuletzt verarbeiteten Element (definiert durch den Status des Lease-Containers) und der neuesten Änderung im Container und überträgt diesen Wert an einen Delegierten. Das Intervall, in dem die Messung durchgeführt wird, kann auch mit einem Standardwert von 5 Sekunden angepasst werden.

Beispiel: Wenn Ihr Änderungsfeedprozessor den aktuellen Versionsmodus verwendet und wie folgt definiert ist:

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

Die richtige Methode, um einen Schätzer zu initialisieren, der den Prozessor misst, wäre die Verwendung von GetChangeFeedEstimatorBuilder wie folgt.

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

Wo sowohl der Prozessor als auch der Schätzer denselben leaseContainer und denselben Namen haben.

Die anderen beiden Parameter sind der Delegat, der eine Zahl empfängt, die angibt, wie viele Änderungen vom Prozessor gelesen werden müssen, und das Zeitintervall, zu dem diese Messung durchgeführt werden soll.

Ein Beispiel für einen Delegaten, der die Schätzung erhält, ist:

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

Sie können diese Schätzung an Ihre Überwachungslösung senden und sie verwenden, um zu verstehen, wie Sich Ihr Fortschritt im Laufe der Zeit verhält.

Detaillierte Schätzung bei Bedarf

Im Gegensatz zum Pushmodell gibt es eine Alternative, mit der Sie die Schätzung nach Bedarf erhalten können. Dieses Modell enthält auch ausführlichere Informationen:

• Die geschätzte Verzögerung pro Mietvertrag.
• Die Instanz, die jede Lease besitzt und verarbeitet, sodass Sie feststellen können, ob ein Problem bei einer Instanz aufgetreten ist.

Wenn Ihr Änderungsfeedprozessor wie folgt definiert ist:

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

Sie können den Schätzer mit derselben Leasekonfiguration erstellen:

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

Und wann immer Sie möchten, mit der von Ihnen benötigten Häufigkeit, können Sie die detaillierte Schätzung erhalten:

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

Jeder ChangeFeedProcessorState enthält die Leasing- und Verzögerungsinformationen und auch welche die aktuelle Instanz ist, die diese Informationen besitzt.

Bereitstellung von Estimator

Der Änderungsfeed-Schätzer muss nicht als Teil des Änderungsfeed-Prozessors bereitgestellt werden oder Teil desselben Projekts sein. Es wird empfohlen, den Schätzwert auf einer unabhängigen Instanz von Ihren Prozessoren bereitzustellen. Eine einzelne Schätzerinstanz kann den Fortschritt für alle Leases und Instanzen in der Bereitstellung des Change Feed Processors nachverfolgen.

Jede Schätzung verbraucht Anforderungseinheiten aus Ihren überwachten Containern und Leasecontainern. Eine Häufigkeit von 1 Minute dazwischen ist ein guter Ausgangspunkt, je niedriger die Häufigkeit, desto höher die verbrauchten Anforderungseinheiten.

In diesem Beispiel wird der Änderungsfeedprozessor im aktuellen Versionsmodus mit der Schätzung der Verzögerung bei der Verarbeitung von Änderungsfeedereignissen verwendet. In der Anwendung werden Dokumente in einen Container (den "Feedcontainer") eingefügt, und gleichzeitig ruft ein anderer Worker-Thread oder eine Worker-Anwendung die eingefügten Dokumente aus dem Change-Feed des Feedcontainers ab und arbeitet auf irgendeine Weise damit.

Der Änderungsfeedprozessor wird wie folgt erstellt und gestartet:

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;
}

Die Überprüfung der Änderungsfeed-Prozessorverzögerung kann von einer separaten Anwendung (z. B. einem Integritätsmonitor) durchgeführt werden, solange dieselben Eingabecontainer (feedContainer und leaseContainer) und dasselbe Leasepräfix (CONTAINER_NAME + "-lease") verwendet werden. Der Schätzcode erfordert, dass der Änderungsfeedprozessor die Möglichkeit hatte, die Dokumente des leaseContainers vollständig zu initialisieren.

Der Schätzer berechnet die akkumulierte Verzögerung, indem er den aktuellen Status des Änderungsfeedprozessors abruft und die geschätzten Verzögerungswerte für jedes verarbeitete Ereignis summiert.

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

Der Gesamtabstand sollte anfangs null sein und schließlich größer oder gleich der Anzahl der erstellten Dokumente sein. Der Gesamtabstandswert kann zur weiteren Analyse protokolliert oder verwendet werden, sodass die Leistung der Änderungsfeedverarbeitung überwacht und potenzielle Engpässe oder Verzögerungen im System identifiziert werden können:

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

Ein Beispiel für einen Delegate, der Änderungen empfängt und diese mit einer Verzögerung behandelt, ist:

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