Korzystanie z narzędzia do szacowania zestawienia zmian

DOTYCZY: NoSQL

W tym artykule opisano, jak można monitorować postęp wystąpień procesora zestawienia zmian podczas odczytywania zestawienia zmian.

Dlaczego postęp monitorowania jest ważny?

Procesor zestawienia zmian działa jako wskaźnik, który przechodzi do przodu w kanale zmian i dostarcza zmiany do implementacji delegata.

Wdrożenie procesora zestawienia zmian może przetwarzać zmiany w określonej szybkości na podstawie dostępnych zasobów, takich jak procesor CPU, pamięć, sieć itd.

Jeśli ta szybkość jest niższa niż szybkość, w jakiej zmiany występują w kontenerze usługi Azure Cosmos DB, procesor zacznie się opóźniać.

Zidentyfikowanie tego scenariusza pomaga zrozumieć, czy musimy skalować wdrożenie procesora zestawienia zmian.

Implementowanie narzędzia do szacowania zestawienia zmian

.NET

Jako model wypychania dla powiadomień automatycznych

Podobnie jak procesor zestawienia zmian, narzędzie do szacowania zestawienia zmian może działać jako model wypychania. Narzędzie do szacowania będzie mierzyć różnicę między ostatnim przetworzonym elementem (zdefiniowanym przez stan kontenera dzierżaw) i najnowszą zmianą w kontenerze, a następnie wypchnie tę wartość do delegata. Interwał, w którym pomiar jest wykonywany, można również dostosować z wartością domyślną wynoszącą 5 sekund.

Jeśli na przykład procesor zestawienia zmian jest zdefiniowany w następujący sposób:

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

Prawidłowy sposób inicjowania narzędzia do szacowania w celu mierzenia tego procesora będzie używany GetChangeFeedEstimatorBuilder w następujący sposób:

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

Gdzie procesor i narzędzie do szacowania mają taką samą leaseContainer i taką samą nazwę.

Pozostałe dwa parametry to delegat, który otrzyma liczbę, która reprezentuje liczbę oczekujących zmian do odczytania przez procesor, oraz przedział czasu, w którym ma zostać podjęta ta miara.

Przykładem delegata, który otrzymuje oszacowanie, jest:

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

Możesz wysłać to oszacowanie do rozwiązania do monitorowania i użyć go, aby zrozumieć, jak postęp działa w czasie.

Jako szczegółowe szacowanie na żądanie

W przeciwieństwie do modelu wypychania istnieje alternatywa umożliwiająca uzyskanie oszacowania na żądanie. Ten model zawiera również bardziej szczegółowe informacje:

• Szacowane opóźnienie na dzierżawę.
• Wystąpienie jest właścicielem i przetwarzaniem każdej dzierżawy, aby określić, czy występuje problem w wystąpieniu.

Jeśli procesor zestawienia zmian jest zdefiniowany w następujący sposób:

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

Narzędzie do szacowania można utworzyć przy użyciu tej samej konfiguracji dzierżawy:

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

I za każdym razem, gdy chcesz, z wymaganą częstotliwością, możesz uzyskać szczegółowe oszacowanie:

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żda ChangeFeedProcessorState z nich będzie zawierać informacje o dzierżawie i opóźnieniu, a także o tym, kto jest jego właścicielem.

Wdrażanie narzędzia do szacowania

Narzędzie do szacowania zestawienia zmian nie musi być wdrażane jako część procesora zestawienia zmian ani być częścią tego samego projektu. Zalecamy wdrożenie narzędzia do szacowania na niezależnym i zupełnie innym wystąpieniu niż procesory. Pojedyncze wystąpienie narzędzia do szacowania może śledzić postęp wszystkich dzierżaw i wystąpień we wdrożeniu procesora zestawienia zmian.

Każde oszacowanie będzie zużywać jednostki żądań z monitorowanych kontenerów i dzierżaw. Częstotliwość 1 minut między jest dobrym punktem wyjścia, im niższa częstotliwość, tym wyższa liczba jednostek żądania zużywanych.

Java

Podany przykład reprezentuje przykładową aplikację Java, która demonstruje implementację procesora zestawienia zmian z oszacowaniem opóźnienia w przetwarzaniu zdarzeń zestawienia zmian. W aplikacji — dokumenty są wstawiane do jednego kontenera (kontener kanału informacyjnego), a w międzyczasie inny wątek roboczy lub aplikacja procesu roboczego ściąga dokumenty wstawione z zestawienia zmian kontenera kanału informacyjnego i działa na nich w jakiś sposób.

Procesor zestawienia zmian został skompilowany i uruchomiony w następujący sposób:

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

Sprawdzanie opóźnienia procesora zestawienia zmian można wykonać w oddzielnej aplikacji (np. monitor kondycji), o ile są używane te same kontenery wejściowe (feedContainer i leaseContainer) oraz dokładnie ten sam prefiks dzierżawy (CONTAINER_NAME + "-lease"). Kod narzędzia do szacowania wymaga, aby procesor zestawienia zmian miał możliwość pełnego zainicjowania dokumentów dzierżawyContainer.

Narzędzie do szacowania oblicza skumulowane opóźnienie, pobierając bieżący stan procesora zestawienia zmian i sumując szacowane wartości opóźnień dla każdego przetwarzanego zdarzenia:

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

Łączne opóźnienie początkowo powinno wynosić zero, a ostatecznie powinno być większe lub równe liczbie utworzonych dokumentów. Łączna wartość opóźnienia może być rejestrowana lub używana do dalszej analizy, umożliwiając monitorowanie wydajności przetwarzania zestawienia zmian i identyfikowanie potencjalnych wąskich gardeł lub opóźnień w systemie:

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

Przykładem delegata, który odbiera zmiany i obsługuje je z opóźnieniem, jest:

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

Dodatkowe zasoby

• Azure Cosmos DB SDK
• Przykłady użycia w witrynie GitHub (.NET)
• Przykłady użycia w witrynie GitHub (Java)
• Dodatkowe przykłady w witrynie GitHub

Następne kroki

Teraz możesz dowiedzieć się więcej o procesorze zestawienia zmian w następujących artykułach:

• Omówienie procesora zestawienia zmian