Använda ändringsflödesestimatorn

GÄLLER FÖR: NoSQL

Den här artikeln beskriver hur du kan övervaka förloppet för dina instanser av ändringsflödesprocessorn när de läser ändringsflödet.

Varför är det viktigt att övervaka förloppet?

Ändringsflödesprocessorn fungerar som en pekare som går framåt i ändringsflödet och levererar ändringarna till en delegatimplementering.

Distributionen av ändringsflödesprocessorn kan bearbeta ändringar med en viss hastighet baserat på dess tillgängliga resurser, till exempel CPU, minne, nätverk och så vidare.

Om den här hastigheten är långsammare än den hastighet med vilken dina ändringar sker i Azure Cosmos DB-containern börjar processorn släpa efter.

Att identifiera det här scenariot hjälper dig att förstå om vi behöver skala distributionen av ändringsflödesprocessorn.

Implementera ändringsflödesestimatorn

.NET

Som push-modell för automatiska meddelanden

Precis som ändringsflödesprocessorn kan ändringsflödesberäknaren fungera som en push-modell. Skattaren mäter skillnaden mellan det senast bearbetade objektet (definierat av tillståndet för lånecontainern) och den senaste ändringen i containern och skickar det här värdet till ett ombud. Det intervall med vilket mätningen utförs kan också anpassas med ett standardvärde på 5 sekunder.

Om ändringsflödesprocessorn till exempel använder det senaste versionsläget och definieras så här:

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

Rätt sätt att initiera en uppskattning för att mäta att processorn skulle använda GetChangeFeedEstimatorBuilder så här:

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

Där både processorn och skattaren delar samma leaseContainer och samma namn.

De andra två parametrarna är ombudet, som tar emot ett tal som representerar hur många ändringar som väntar på att läsas av processorn och det tidsintervall som du vill att mätningen ska utföras med.

Ett exempel på ett ombud som tar emot uppskattningen är:

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

Du kan skicka den här uppskattningen till din övervakningslösning och använda den för att förstå hur dina framsteg fungerar över tid.

Som en detaljerad uppskattning på begäran

Till skillnad från push-modellen finns det ett alternativ som gör att du kan hämta uppskattningen på begäran. Den här modellen innehåller också mer detaljerad information:

• Den uppskattade fördröjningen per lån.
• Instansen äger och bearbetar varje lån, så att du kan identifiera om det finns ett problem på en instans.

Om ändringsflödesprocessorn definieras så här:

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

Du kan skapa uppskattningen med samma lånekonfiguration:

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

Och när du vill ha det, med den frekvens du behöver, kan du få den detaljerade uppskattningen:

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

Var ChangeFeedProcessorState och en innehåller låne- och fördröjningsinformationen, samt vem som är den aktuella instansen som äger den.

Beräknardistribution

Ändringsflödesberäknaren behöver inte distribueras som en del av ändringsflödesprocessorn eller vara en del av samma projekt. Vi rekommenderar att du distribuerar uppskattningen på en oberoende instans från dina processorer. En enskild beräkningsinstans kan spåra förloppet för alla lån och instanser i distributionen av ändringsflödesprocessorn.

Varje uppskattning förbrukar enheter för begäranden från dina övervakade containrar och lånecontainrar. En frekvens på 1 minut mellan är en bra startpunkt, desto lägre frekvens, desto högre förbrukas enheter för begäran.

Java

I det här exemplet används ändringsflödesprocessorn i senaste versionsläge med uppskattningen av fördröjningen i bearbetningen av ändringsflödeshändelser. I programmet - dokument infogas i en container ("feedcontainern") och under tiden hämtar en annan arbetstråd eller arbetsprogram infogade dokument från feedcontainerns ändringsflöde och arbetar på dem på något sätt.

Ändringsflödesprocessorn skapas och startas så här:

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

Fördröjningskontroll av ändringsflödesprocessor kan utföras i ett separat program (till exempel en hälsoövervakare) så länge samma indatacontainer (feedContainer och leaseContainer) och exakt samma låneprefix (CONTAINER_NAME + "-lease") används. Beräkningskoden kräver att ändringsflödesprocessorn hade möjlighet att helt initiera leaseContainer-dokumenten.

Skattaren beräknar den ackumulerade fördröjningen genom att hämta ändringsflödesprocessorns aktuella tillstånd och summera de uppskattade fördröjningsvärdena för varje händelse som bearbetas:

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

Den totala fördröjningen bör initialt vara noll och bör slutligen vara större eller lika med antalet dokument som skapas. Det totala fördröjningsvärdet kan loggas eller användas för ytterligare analys, vilket gör det möjligt att övervaka prestanda för ändringsflödesbearbetningen och identifiera eventuella flaskhalsar eller fördröjningar i systemet:

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

Ett exempel på ett ombud som tar emot ändringar och hanterar dem med en fördröjning är:

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

Ändringsflödeslägen som stöds

Ändringsflödesberäknaren kan användas för både det senaste versionsläget och för alla versioner och borttagningsläge. I båda lägena är den angivna uppskattningen inte garanterad att vara ett exakt antal utestående ändringar att bearbeta.

Ytterligare resurser

• Azure Cosmos DB SDK
• Användningsexempel på GitHub (.NET senaste version)
• Användningsexempel på GitHub (.NET alla versioner och borttagningar)
• Användningsexempel på GitHub (Java)
• Ytterligare exempel på GitHub

Nästa steg

Du kan nu fortsätta med att lära dig mer om ändringsflödesprocessorn i följande artikel:

• Översikt över ändringsflödesprocessor