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

Bu makalede değişiklik akışı işlemcisi örneklerinizin değişiklik akışını okurken ilerlemesini nasıl izleyebileceğ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 push modeli olarak

Değişiklik akışı işlemcisi gibi, değişiklik akışı tahmin aracı da itme 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çer ve bu değeri bir temsilciye iletir. Ö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 en son sürüm modunu kullanıyorsa ve şöyle 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();

İşlemciyi ölçmek için bir tahminci başlatmanın doğru yolu, şu şekilde GetChangeFeedEstimatorBuilder kullanmaktır:

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.

İşlemci tarafından okunmayı bekleyen kaç değişiklik olduğunu belirten sayı alan bir delege ve bu ölçümün alınmasını istediğiniz zaman aralığı, diğer iki parametredir.

Bir tahmini alan temsilciye örnek:

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

İtme modelinin aksine, tahmini talep üzerine 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 kirayı sahiplenen ve işleyen örneği, bir örnekte sorun olup olmadığını belirleyebilmeniz için tanımlayabilirsiniz.

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

Tahminciyi 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 ayrıca bu bilgilere sahip olan geçerli örneğin kim olduğunu içerir.

Estimatör 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 bir örneğe dağıtmanızı öneririz. Tek bir tahminleyici örneği, değişim akışı işlemcisi dağıtımınızdaki tüm kiralamalar ve örneklerin ilerleme durumunu izleyebilir.

Her tahmin, izlenen ve kira kapsayıcılarınızdakiistek 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

Bu örnek, değişiklik akışı olaylarını işlemedeki gecikmenin tahmini ile en son sürüm modunda değişiklik akışı işlemcisini kullanır. Uygulamada - belgeler bir kapsayıcıya ("besleme kapsayıcısı") ekleniyor ve aynı zamanda başka bir çalışan iş parçacığı veya çalışan uygulaması, eklenen belgeleri besleme 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;
}

Değişiklik akışı işlemcisinin gecikme denetimi, aynı girdi konteynerleri (feedContainer ve leaseContainer) ve tamamen aynı kira ön eki (CONTAINER_NAME + "-lease") kullanıldığı sürece ayrı bir uygulamada (sistem durumu izleyicisi gibi) gerçekleştirilebilir. Tahmin kodu, değişiklik akışı işlemcisinin leaseContainer belgelerini tam olarak başlatma fırsatına sahip olmasını gerektirir.

Tahmin aracı, değişiklik akışı işlemcisinin 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ı gecikmeli olarak 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()");
    };
}

Desteklenen değişiklik akışı modları

Değişiklik akışı tahmin aracı hem en son sürüm modu hem de tüm sürümler ve silmeler modu için kullanılabilir. Her iki modda da sağlanan tahminin, işlenmesi gereken bekleyen değişikliklerin tam sayısı olduğu garanti değildir.

Ek kaynaklar

• Azure Cosmos DB SDK'sı
• GitHub'da kullanım örnekleri (.NET en son sürümü)
• GitHub'da kullanım örnekleri (.NET tüm sürümler ve silmeler)
• GitHub'da kullanım örnekleri (Java)
• GitHub'da ek örnekler

Sonraki Adımlar

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

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