Použití estimátoru kanálu změn

PLATÍ PRO: NoSQL

Tento článek popisuje, jak můžete sledovat průběh instancí procesoru kanálu změn při čtení kanálu změn.

Proč je monitorování průběhu důležité?

Procesor kanálu změn funguje jako ukazatel, který se posune vpřed v kanálu změn a doručí změny implementaci delegáta.

Nasazení procesoru kanálu změn může zpracovávat změny určitým tempem na základě dostupných prostředků, jako je procesor, paměť, síť atd.

Pokud je tato rychlost pomalejší než rychlost, s jakou se změny projeví v kontejneru Azure Cosmos DB, procesor se začne zaostávat.

Identifikace tohoto scénáře pomáhá pochopit, jestli potřebujeme škálovat nasazení procesoru kanálu změn.

Implementace estimátoru kanálu změn

.NET

Jako model nabízených oznámení pro automatická oznámení

Podobně jako procesor kanálu změn může estimátor kanálu změn fungovat jako model nabízených oznámení. Estimátor změří rozdíl mezi poslední zpracovávanou položkou (definovanou stavem kontejneru zapůjčení) a nejnovější změnou v kontejneru a nasdílí tuto hodnotu delegátu. Interval, ve kterém je měření pořízeno, lze také přizpůsobit výchozí hodnotou 5 sekund.

Například pokud je procesor kanálu změn definovaný takto:

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

Správný způsob inicializace odhadce pro měření procesoru by vypadal GetChangeFeedEstimatorBuilder takto:

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

Kde procesor i odhadce sdílejí stejný leaseContainer a stejný název.

Další dva parametry jsou delegát, který obdrží číslo, které představuje , kolik změn čeká na přečtení procesorem, a časový interval, ve kterém chcete toto měření vzít.

Příkladem delegáta, který obdrží odhad, je:

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

Tento odhad můžete odeslat do svého monitorovacího řešení a použít ho k pochopení toho, jak se váš pokrok v průběhu času chová.

Podrobný odhad na vyžádání

Na rozdíl od modelu nabízení existuje alternativa, která umožňuje získat odhad na vyžádání. Tento model také poskytuje podrobnější informace:

• Odhadovaná prodleva na zapůjčení
• Instance, která vlastní a zpracovává každé zapůjčení, takže můžete zjistit, jestli u instance došlo k problému.

Pokud je procesor kanálu změn definovaný takto:

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

Odhadce můžete vytvořit se stejnou konfigurací zapůjčení:

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

A kdykoli to budete chtít, s požadovanou frekvencí, můžete získat podrobný odhad:

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

Každý z nich ChangeFeedProcessorState bude obsahovat informace o zapůjčení a prodlevě a také to, kdo je aktuální instancí, která ji vlastní.

Nasazení estimátoru

Estimátor kanálu změn nemusí být nasazen jako součást procesoru kanálu změn ani nemusí být součástí stejného projektu. Doporučujeme nasadit estimátor na nezávislou a úplně jinou instanci než procesory. Jedna instance estimátoru může sledovat průběh všech zapůjčení a instancí v nasazení procesoru kanálu změn.

Každý odhad bude spotřebovávat jednotky žádostí z monitorovaných kontejnerů a kontejnerů zapůjčení. Frekvence mezi 1 minutou je dobrým výchozím bodem, čím nižší je frekvence, tím vyšší spotřebované jednotky žádostí.

Java

Uvedený příklad představuje ukázkovou aplikaci Java, která demonstruje implementaci procesoru kanálu změn s odhadem prodlevy při zpracování událostí kanálu změn. V aplikaci se dokumenty vkládají do jednoho kontejneru (kontejner informačního kanálu) a mezitím další pracovní vlákno nebo pracovní aplikace načítá vložené dokumenty z kanálu kanálu Change Feed a pracuje na nich nějakým způsobem.

Procesor kanálu změn se sestaví a spustí takto:

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

Kontrolu prodlevy procesoru kanálu změn je možné provádět v samostatné aplikaci (jako je monitor stavu), pokud se použijí stejné vstupní kontejnery (feedContainer a leaseContainer) a stejná předpona zapůjčení (CONTAINER_NAME + "-lease"). Kód estimátoru vyžaduje, aby procesor kanálu změn měl příležitost plně inicializovat dokumenty leaseContainer.

Estimátor vypočítá kumulovanou prodlevu načtením aktuálního stavu procesoru kanálu změn a sečtováním odhadovaných hodnot prodlevy pro každou zpracovávanou událost:

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

Celková prodleva by zpočátku měla být nula a nakonec by měla být větší nebo rovna počtu vytvořených dokumentů. Celkovou hodnotu prodlevy lze protokolovat nebo použít k další analýze, což umožňuje monitorovat výkon zpracování kanálu změn a identifikovat případné kritické body nebo zpoždění v systému:

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

Příkladem delegáta, který přijímá změny a zpracovává je s prodlevou, je:

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

Další materiály

• Azure Cosmos DB SDK
• Ukázky využití na GitHubu (.NET)
• Ukázky využití na GitHubu (Java)
• Další ukázky na GitHubu

Další kroky

Teď můžete pokračovat a získat další informace o procesoru kanálu změn v následujících článcích:

• Přehled procesoru kanálu změn