Değişiklik akışı tahmin aracını kullanma

UYGULANANLAR: NoSQL

Bu makalede, değişiklik akışı işlemcisi örneklerinizin değişiklik akışını okurken ilerleme durumunu nasıl izleyebildiğiniz açıklanmaktadır.

İzleme ilerlemesi neden önemlidir?

Değişiklik akışı işlemcisi, değişiklik akışınızda ileriye doğru hareket eden ve değişiklikleri temsilci uygulamasına teslim eden bir işaretçi işlevi görür.

Değişiklik akışı işlemcisi dağıtımınız CPU, bellek, ağ gibi kullanılabilir kaynaklarına göre değişiklikleri belirli bir hızda işleyebilir.

Bu hız, Değişikliklerinizin Azure Cosmos DB kapsayıcınızda gerçekleşme hızından daha yavaşsa işlemciniz gecikmeye başlar.

Bu senaryoyı tanımlamak, değişiklik akışı işlemci dağıtımımızı ölçeklendirmemiz gerekip gerekmediğini anlamanıza yardımcı olur.

Değişiklik akışı tahmin aracını uygulama

.NET

Otomatik bildirimler için anında iletme modeli olarak

Değişiklik akışı işlemcisi gibi, değişiklik akışı tahmin aracı da anında iletme modeli olarak çalışabilir. Tahmin aracı, son işlenen öğe (kira kapsayıcısının durumuyla tanımlanır) ile kapsayıcıdaki en son değişiklik arasındaki farkı ölçecek ve bu değeri bir temsilciye gönderecektir. Ölçümün alındığı aralık, varsayılan 5 saniyelik bir değerle de özelleştirilebilir.

Örneğin, değişiklik akışı işlemciniz şu şekilde tanımlanmışsa:

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

İşlemcinin şu şekilde kullanacağını GetChangeFeedEstimatorBuilder ölçmek için tahmin aracı başlatmanın doğru yolu:

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

Burada hem işlemci hem de tahmin aracı aynı leaseContainer ve aynı adı paylaşır.

Diğer iki parametre, işlemci tarafından kaç değişikliğin okunmasını beklediğini ve bu ölçümün alınmasını istediğiniz zaman aralığını gösteren bir sayı alan temsilcidir.

Tahmini alan bir temsilci örneği:

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

Bu tahmini izleme çözümünüze gönderebilir ve ilerleme durumunuzun zaman içinde nasıl çalıştığını anlamak için kullanabilirsiniz.

İsteğe bağlı ayrıntılı tahmin olarak

Gönderme modelinin aksine, tahminleri isteğe bağlı olarak elde etmenizi sağlayan bir alternatif vardır. Bu model ayrıca daha ayrıntılı bilgiler sağlar:

• Kira başına tahmini gecikme.
• Her kiraya sahip olan ve işleyen örnek, bir örnekte sorun olup olmadığını belirleyebilirsiniz.

Değişiklik akışı işlemciniz şu şekilde tanımlanmışsa:

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

Tahmin aracını aynı kira yapılandırmasıyla oluşturabilirsiniz:

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

İstediğiniz zaman, ihtiyacınız olan sıklıkta ayrıntılı tahmin elde edebilirsiniz:

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

Her ChangeFeedProcessorState biri kiralama ve gecikme bilgilerini ve aynı zamanda buna sahip olan geçerli örneğin kim olduğunu içerir.

Tahmin aracı dağıtımı

Değişiklik akışı tahmin aracının, değişiklik akışı işlemcinizin bir parçası olarak dağıtılması veya aynı projenin parçası olması gerekmez. Tahmin aracını işlemcilerinizden bağımsız ve tamamen farklı bir örneğe dağıtmanızı öneririz. Tek bir tahmin örneği, değişiklik akışı işlemcisi dağıtımınızdaki tüm kiraların ve örneklerin ilerleme durumunu izleyebilir.

Her tahmin, izlenen ve kira kapsayıcılarınızdaki istek birimlerini kullanır. Aralarında 1 dakikalık bir sıklık iyi bir başlangıç noktasıdır, sıklık ne kadar düşük olursa, kullanılan istek birimleri o kadar yüksek olur.

Java

Sağlanan örnek, değişiklik akışı olaylarını işlemedeki gecikmenin tahminiyle Değişiklik Akışı İşlemcisi'nin uygulanmasını gösteren örnek bir Java uygulamasını temsil eder. Uygulamada - belgeler bir kapsayıcıya ("akış kapsayıcısı") ekleniyor ve bu arada başka bir çalışan iş parçacığı veya çalışan uygulaması, eklenen belgeleri akış kapsayıcısının Değişiklik Akışı'ndan çekiyor ve bunlar üzerinde bir şekilde çalışıyor.

Değişiklik Akışı İşlemcisi şu şekilde oluşturulur ve başlatılır:

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

Aynı giriş kapsayıcıları (feedContainer ve leaseContainer) ve tam olarak aynı kira ön eki () kullanıldığı sürece Değişiklik Akışı İşlemcisi gecikme denetimi ayrı bir uygulamada (CONTAINER_NAME + "-lease"sistem durumu izleyicisi gibi) gerçekleştirilebilir. Tahmin kodu, Değişiklik Akışı İşlemcisi'nin leaseContainer belgelerini tam olarak başlatma fırsatına sahip olmasını gerektirir.

Tahmin aracı, Değişiklik Akışı İşlemcisi'nin geçerli durumunu alarak ve işlenen her olay için tahmini gecikme değerlerini toplayarak birikmiş gecikmeyi hesaplar:

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

Toplam gecikme başlangıçta sıfır olmalı ve son olarak da oluşturulan belge sayısına eşit veya daha büyük olmalıdır. Toplam gecikme değeri günlüğe kaydedilebilir veya daha fazla analiz için kullanılabilir, böylece değişiklik akışı işleme performansını izleyebilir ve sistemdeki olası performans sorunlarını veya gecikmeleri belirleyebilirsiniz:

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

Değişiklikleri alan ve bunları gecikme süresiyle işleyen bir temsilci örneği:

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

Ek kaynaklar

• Azure Cosmos DB SDK'sı
• GitHub'da kullanım örnekleri (.NET)
• GitHub'da kullanım örnekleri (Java)
• GitHub'da ek örnekler

Sonraki adımlar

Şimdi aşağıdaki makalelerde değişiklik akışı işlemcisi hakkında daha fazla bilgi edinebilirsiniz:

• Değişiklik akışı işlemcisine genel bakış