De estimator van de wijzigingenfeed gebruiken

VAN TOEPASSING OP: NoSQL

In dit artikel wordt beschreven hoe u de voortgang van de processorexemplaren van de wijzigingenfeed kunt controleren terwijl ze de wijzigingenfeed lezen.

Waarom is het bewaken van de voortgang belangrijk?

De wijzigingenfeedprocessor fungeert als een aanwijzer die door uw wijzigingenfeed wordt verplaatst en de wijzigingen aan een gedelegeerde implementatie levert.

De implementatie van de wijzigingenfeedprocessor kan wijzigingen met een bepaalde snelheid verwerken op basis van de beschikbare resources, zoals CPU, geheugen, netwerk, enzovoort.

Als deze snelheid langzamer is dan de snelheid waarmee uw wijzigingen in uw Azure Cosmos DB-container plaatsvinden, begint uw processor achter te blijven.

Als u dit scenario identificeert, begrijpt u of we de implementatie van de wijzigingenfeedprocessor moeten schalen.

De wijzigingsfeed-estimator implementeren

.NET

Als pushmodel voor automatische meldingen

Net als de verwerker van de wijzigingenfeed kan de wijzigingsfeedschatter werken als pushmodel. De estimator meet het verschil tussen het laatst verwerkte item (gedefinieerd door de status van de leasecontainer) en de meest recente wijziging in de container en pusht deze waarde naar een gemachtigde. Het interval waarmee de meting wordt uitgevoerd, kan ook worden aangepast met een standaardwaarde van 5 seconden.

Als uw wijzigingenfeedprocessor bijvoorbeeld de nieuwste versiemodus gebruikt en als volgt wordt gedefinieerd:

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

De juiste manier om een estimator te initialiseren om die processor te meten, gebruikt GetChangeFeedEstimatorBuilder als volgt:

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

Waarbij zowel de processor als de estimator dezelfde leaseContainer en dezelfde naam hebben.

De andere twee parameters zijn de gedelegeerde, die een getal ontvangt dat aangeeft hoeveel wijzigingen moeten worden gelezen door de processor en het tijdsinterval waarop u deze meting wilt uitvoeren.

Een voorbeeld van een gemachtigde die de schatting ontvangt, is:

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

U kunt deze schatting naar uw bewakingsoplossing verzenden en deze gebruiken om te begrijpen hoe uw voortgang zich na verloop van tijd gedraagt.

Als een gedetailleerde schatting op aanvraag

In tegenstelling tot het pushmodel is er een alternatief waarmee u de schatting op aanvraag kunt verkrijgen. Dit model biedt ook meer gedetailleerde informatie:

• De geschatte vertraging per lease.
• Het exemplaar dat eigenaar is van en elke lease verwerkt, zodat u kunt vaststellen of er een probleem is op een exemplaar.

Als uw wijzigingenfeedprocessor als volgt is gedefinieerd:

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

U kunt de estimator maken met dezelfde leaseconfiguratie:

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

En wanneer u wilt, met de frequentie die u nodig hebt, kunt u de gedetailleerde schatting verkrijgen:

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

Elk ChangeFeedProcessorState bevat de lease- en vertragingsgegevens en ook de huidige instantie die eigenaar is van het huidige exemplaar.

Implementatie van estimator

De wijzigingsfeed-estimator hoeft niet te worden geïmplementeerd als onderdeel van uw wijzigingenfeedverwerker en hoeft ook geen deel uit te maken van hetzelfde project. U wordt aangeraden de estimator te implementeren op een onafhankelijk exemplaar van uw processors. Eén estimator-exemplaar kan de voortgang van alle leases en exemplaren in de implementatie van de wijzigingenfeedprocessor bijhouden.

Elke schatting verbruikt aanvraageenheden uit uw bewaakte en leasecontainers. Een frequentie van 1 minuut tussen is een goed startpunt, hoe lager de frequentie, hoe hoger de verbruikte aanvraageenheden.

Java

In dit voorbeeld wordt de processor van de wijzigingenfeed gebruikt in de nieuwste versiemodus met de schatting van de vertraging in het verwerken van wijzigingenfeed-gebeurtenissen. In de toepassing worden documenten ingevoegd in één container (de 'feedcontainer'),) en ondertussen worden in een andere werkthread of werkroltoepassing ingevoegde documenten opgehaald uit de wijzigingenfeed van de feedcontainer en op een of andere manier uitgevoerd.

De wijzigingenfeedprocessor is gebouwd en gestart als volgt:

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

De vertragingscontrole van de wijzigingenfeedverwerker kan worden uitgevoerd op een afzonderlijke toepassing (zoals een statuscontrole) zolang dezelfde invoercontainers (feedContainer en leaseContainer) en exact hetzelfde leasevoorvoegsel (CONTAINER_NAME + "-lease") worden gebruikt. De estimatorcode vereist dat de verwerker van de wijzigingenfeed de mogelijkheid heeft gehad om de documenten van de leaseContainer volledig te initialiseren.

De estimator berekent de samengevoegde vertraging door de huidige status van de wijzigingenfeedprocessor op te halen en de geschatte vertragingswaarden op te tellen voor elke gebeurtenis die wordt verwerkt:

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

De totale vertraging moet in eerste instantie nul zijn en ten slotte groter of gelijk zijn aan het aantal gemaakte documenten. De totale vertragingswaarde kan worden geregistreerd of gebruikt voor verdere analyse, zodat de prestaties van de verwerking van de wijzigingenfeed kunnen worden bewaakt en mogelijke knelpunten of vertragingen in het systeem kunnen worden geïdentificeerd:

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

Een voorbeeld van een gemachtigde die wijzigingen ontvangt en deze met een vertraging verwerkt, is:

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

Ondersteunde modi voor wijzigingenfeeds

De estimator van de wijzigingenfeed kan worden gebruikt voor zowel de nieuwste versiemodus als alle versies en de modus Voor verwijderen. In beide modi is de opgegeven schatting niet gegarandeerd een exacte telling van openstaande wijzigingen die moeten worden verwerkt.

Aanvullende bronnen

• Azure Cosmos DB SDK
• Gebruiksvoorbeelden op GitHub (nieuwste versie van.NET)
• Gebruiksvoorbeelden op GitHub (.NET alle versies en verwijderingen)
• Gebruiksvoorbeelden op GitHub (Java)
• Aanvullende voorbeelden op GitHub

Volgende stappen

U kunt nu verdergaan met meer informatie over de verwerker van wijzigingenfeeds in het volgende artikel:

• Overzicht van wijzigingenfeedprocessor