Használja a változáscsatorna-becslőt

Ez a cikk azt ismerteti, hogyan figyelheti a változáscsatorna processzorpéldányainak állapotát a változáscsatorna olvasása közben.

Miért fontos az előrehaladás monitorozása?

A változáscsatorna feldolgozója olyan mutatóként működik, amely előrehalad a változáscsatornán , és a módosításokat egy delegált implementációnak továbbítja.

A változáscsatorna-feldolgozó üzembe helyezése bizonyos sebességgel képes feldolgozni a módosításokat a rendelkezésre álló erőforrások, például a processzor, a memória, a hálózat stb. alapján.

Ha ez a sebesség lassabb, mint az Azure Cosmos DB-tárolóban végrehajtott módosítások sebessége, a processzor kezd lemaradni.

Ennek a forgatókönyvnek a azonosítása segít megérteni, hogy szükség van-e a változáscsatorna processzor üzembe helyezésének skálázására.

A változáscsatorna-becslő implementálása

.NET

Push modellként automatikus értesítések számára

A változáscsatorna-feldolgozóhoz hasonlóan a változáscsatorna-becslő leküldéses modellként is működhet. A becslő méri az utolsó feldolgozott elem (a kölcsönzési tároló állapota által meghatározva) és a tároló legutóbbi változása közötti különbséget, és ezt az értéket továbbítja egy delegáltnak. A mérés időköze az alapértelmezett 5 másodperces értékkel is testre szabható.

Példaként, ha a változáscsatorna-feldolgozó a legújabb verziómódot használja, és a következőképpen van definiálva:

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

A processzor méréséhez szükséges becslő objektum helyes inicializálása a következőképpen történik GetChangeFeedEstimatorBuilder felhasználásával:

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

Ahol a feldolgozó és a becslő is ugyanazzal leaseContainer a névvel rendelkezik.

A másik két paraméter a delegált, amely egy számot kap, amely azt jelzi, hogy a processzor hány módosítást vár olvasásra , és azt az időtartamot, amelyen a mérést el szeretné végezni.

Példa a becslést fogadó meghatalmazottra:

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

Ezt a becslést elküldheti a monitorozási megoldásnak, és segítségével megértheti, hogyan működik az előrehaladás az idő múlásával.

Igény szerinti részletes becslésként

A leküldéses modellel ellentétben van egy alternatíva, amely lehetővé teszi a becslés igény szerinti lekérését. Ez a modell részletesebb információkat is nyújt:

• A bérletenkénti becsült késedelmi idő.
• Minden egyes bérlethez tartozik egy példány, amely felel a tulajdonlásért és feldolgozásért, így meg tudja állapítani, hogy van-e probléma az adott példányon.

Ha a változáscsatorna-feldolgozó a következőképpen van definiálva:

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

A becslőt ugyanazzal a bérletkonfigurációval hozhatja létre:

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

És amikor csak szeretné, a szükséges gyakorisággal beszerezheti a részletes becslést:

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

Mindegyik ChangeFeedProcessorState tartalmazza a bérlet és késés adatait, valamint azt is, hogy ki az aktuális példány tulajdonosa.

Becslő üzembe helyezése

A változáscsatorna-becslőt nem kell üzembe helyezni a változáscsatorna-feldolgozó részeként, és nem is kell ugyanabban a projektben részt vennie. Javasoljuk, hogy a becslőt egy független példányon helyezze üzembe a processzoroktól. Egyetlen becslési példány nyomon követheti a változáscsatorna-feldolgozó üzembe helyezésének összes bérletének és példányának állapotát.

Minden becslés kérelemegységeket használ fel a figyelt és bérelt tárolók közül. A kettő közötti 1 perces gyakoriság jó kiindulási pont, minél alacsonyabb a gyakoriság, annál magasabb a felhasznált kérelemegységek száma.

Java

Ez a példa a változáscsatorna-feldolgozót a legújabb verzió módban használja a változáscsatorna-események feldolgozásának késésének becslésével. Az alkalmazásban a dokumentumokat egy tárolóba (az úgynevezett "adatáramlási tárolóba") szúrják be, miközben egy másik feldolgozószál vagy feldolgozó alkalmazás lekéri a beszúrt dokumentumokat az adatáramlási tároló változáscsatornájából, és valamilyen módon feldolgozza őket.

A változási Feed Processzor a következőképpen van felépítve és elindítva:

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

A változási folyamatok késésének ellenőrzése külön alkalmazáson (például egy egészségügyi figyelőrendszeren) végezhető el, amennyiben ugyanazokat a bemeneti tárolókat (feedContainer és leaseContainer) és pontosan ugyanazt a bérletelőtagot (CONTAINER_NAME + "-lease") használják. A becslési kód megköveteli, hogy a change feed processor-nak lehetősége legyen teljes mértékben inicializálni a leaseContainer dokumentumait.

A becslő kiszámítja a halmozott késést a változáscsatorna-feldolgozó aktuális állapotának beolvasásával és az egyes feldolgozott események becsült késési értékeinek összegzésével:

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

A teljes késésnek kezdetben nullának kell lennie, és végül nagyobbnak vagy egyenlőnek kell lennie a létrehozott dokumentumok számával. A teljes késési érték naplózható vagy felhasználható további elemzésekhez, lehetővé téve a változáscsatorna-feldolgozás teljesítményének monitorozását és a rendszer esetleges szűk keresztmetszeteinek vagy késéseinek azonosítását:

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élda egy olyan meghatalmazottra, aki fogadja a módosításokat, és késéssel kezeli őket:

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

Támogatott változáscsatorna-módok

A változáscsatorna-becslés a legújabb verziómódhoz , valamint az összes verzióhoz és törlési módhoz is használható. Mindkét módban a megadott becslés nem garantáltan a kivételesen fennálló változások pontos száma.

További erőforrások

• Azure Cosmos DB SDK
• Használati minták a GitHubon (.NET legújabb verzió)
• Használati minták a GitHubon (.NET minden verzió és törlés)
• Használati példák a GitHubon (Java)
• További minták a GitHubon

Következő lépések

A változáscsatorna-feldolgozóról az alábbi cikkben olvashat bővebben:

• A változásnapló-feldolgozó áttekintése