Použití odhadce zdroje 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 posouvá vpřed ve vašem změnovém kanálu a doručuje změny implementaci delegáta.

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

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

Identifikace tohoto scénáře pomáhá pochopit, jestli potřebujeme škálování nasazení procesoru sledování 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 měří rozdíl mezi poslední zpracovávanou položkou (definovanou stavem kontejneru zapůjčení) a poslední změnou v kontejneru a odešle tuto hodnotu delegátu. Interval, ve kterém je měření pořízeno, lze také přizpůsobit výchozí hodnotou 5 sekund.

Pokud například procesor kanálu změn používá nejnovější režim verze a definuje se 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 č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é zpoždění pro pronájem
• 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ý ChangeFeedProcessorState obsahuje informace o pronájmu a prodlevě a také o tom, která je aktuální instance, která jej 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 instanci z procesorů. Jedna instance odhadovače může sledovat průběh všech pronájmů a instancí v nasazení procesoru sledování změn.

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

Java

Tento příklad používá procesor kanálu změn v nejnovějším režimu verze s odhadem prodlevy při zpracování událostí kanálu změn. V aplikaci se dokumenty vkládají do jednoho kontejneru (tzv. "feed kontejner") a mezitím další pracovní vlákno nebo pracovní aplikace načítá vložené dokumenty ze změnového informačního kanálu a pracuje na nich nějakým způsobem.

Procesor změnového kanálu se vytvoří 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()");
    };
}

Podporované režimy kanálu změn

Nástroj pro odhad kanálu změn se dá použít jak pro režim nejnovější verze, tak pro režim všech verzí a odstranění. V obourežimch

Další materiály

• Azure Cosmos DB SDK
• Ukázky využití na GitHubu (nejnovější verze .NET)
• Ukázky využití na GitHubu (.NET všechny verze a odstranění)
• Ukázky využití na GitHubu (Java)
• Další ukázky na GitHubu

Další kroky

Nyní se můžete dozvědět více o procesoru změnového toku v následujícím článku:

• Přehled nástroje pro zpracování kanálu změn