Gunakan estimator umpan perubahan

BERLAKU UNTUK: NoSQL

Artikel ini menjelaskan bagaimana Anda dapat memantau kemajuan instans prosesor umpan perubahan saat membaca umpan perubahan.

Mengapa memantau kemajuan penting?

Prosesor umpan perubahan bertindak sebagai penunjuk yang bergerak maju di seluruh umpan perubahan Anda dan memberikan perubahan pada implementasi delegasi.

Penyebaran prosesor umpan perubahan Anda dapat memproses perubahan pada tingkat tertentu berdasarkan sumber daya yang tersedia seperti CPU, memori, jaringan, dan sebagainya.

Jika laju ini lebih lambat dari tingkat di mana perubahan Anda terjadi di kontainer Azure Cosmos DB Anda, prosesor Anda akan mulai tertinggal.

Mengidentifikasi skenario ini membantu memahami apakah kita perlu menskalakan penyebaran prosesor umpan perubahan kami.

Mengimplementasikan estimator umpan perubahan

.NET

Sebagai model pendorongan untuk pemberitahuan otomatis

Seperti prosesor umpan perubahan, estimator umpan perubahan dapat berfungsi sebagai model pendorongan. Estimator akan mengukur perbedaan antara item yang diproses terakhir (didefinisikan oleh keadaan kontainer sewa) dan perubahan terbaru dalam kontainer, dan mendorong nilai ini ke delegasi. Interval di mana pengukuran diambil juga dapat disesuaikan dengan nilai default 5 detik.

Sebagai contoh, jika prosesor umpan perubahan Anda didefinisikan seperti ini:

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

Cara yang benar untuk menginisialisasi estimator untuk mengukur prosesor tersebut akan menggunakan GetChangeFeedEstimatorBuilder seperti:

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

Di mana prosesor dan estimator memiliki nama dan leaseContainer yang sama.

Dua parameter lainnya adalah delegasi, yang akan menerima angka yang menunjukkan berapa banyak perubahan yang tertunda untuk dibaca oleh prosesor, dan interval waktu di mana Anda ingin pengukuran ini diambil.

Contoh delegasi yang menerima estimasi adalah:

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

Anda dapat mengirim estimasi ini ke solusi pemantauan Anda dan menggunakannya untuk memahami bagaimana kemajuan Anda berperilaku dari waktu ke waktu.

Sebagai estimasi terperinci sesuai permintaan

Berbeda dengan model pendorongan, ada alternatif yang memungkinkan Anda mendapatkan estimasi sesuai permintaan. Model ini juga memberikan informasi yang lebih detail:

• Perkiraan jeda per sewa.
• Instans yang memiliki dan memproses setiap sewa, sehingga Anda dapat mengidentifikasi apakah ada masalah pada instans.

Jika prosesor umpan perubahan Anda didefinisikan seperti ini:

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

Anda dapat membuat estimator dengan konfigurasi sewa yang sama:

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

Dan kapan pun Anda menginginkannya, dengan frekuensi yang Anda butuhkan, Anda dapat memperoleh estimasi terperinci:

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

Setiap ChangeFeedProcessorState akan berisi informasi sewa dan jeda, dan juga siapa instans saat ini yang memilikinya.

Penyebaran estimator

Estimator umpan perubahan tidak perlu digunakan sebagai bagian dari prosesor umpan perubahan Anda, atau menjadi bagian dari proyek yang sama. Sebaiknya sebarkan estimator pada instans independen dan sama sekali berbeda dari prosesor Anda. Satu instans estimator dapat melacak kemajuan untuk semua sewa dan instans dalam penyebaran prosesor umpan perubahan Anda.

Setiap estimasi akan menggunakan unit permintaan dari kontainer yang dipantau dan disewakan. Frekuensi 1 menit di antaranya adalah titik awal yang baik, semakin rendah frekuensi, semakin tinggi unit permintaan yang dikonsumsi.

Java

Contoh yang disediakan mewakili contoh aplikasi Java yang menunjukkan implementasi Prosesor Umpan Perubahan dengan estimasi jeda dalam memproses peristiwa umpan perubahan. Dalam aplikasi - dokumen sedang dimasukkan ke dalam satu kontainer ("kontainer umpan"), dan sementara itu utas pekerja atau aplikasi pekerja lain menarik dokumen yang disisipkan dari Umpan Perubahan kontainer umpan dan mengoperasikannya dalam beberapa cara.

Prosesor Umpan perubahan dibuat dan dimulai seperti ini:

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

Pemeriksaan lag Prosesor Umpan Perubahan dapat dilakukan pada aplikasi terpisah (seperti monitor kesehatan) selama kontainer input yang sama (feedContainer dan leaseContainer) dan awalan sewa yang sama persis (CONTAINER_NAME + "-lease") digunakan. Kode estimator mengharuskan Prosesor Umpan Perubahan memiliki kesempatan untuk menginisialisasi dokumen leaseContainer sepenuhnya.

Estimator menghitung akumulasi jeda dengan mengambil status Prosesor Umpan Perubahan saat ini dan meringkas perkiraan nilai jeda untuk setiap peristiwa yang sedang diproses:

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

Total jeda awalnya harus nol dan akhirnya harus lebih besar atau sama dengan jumlah dokumen yang dibuat. Nilai total jeda dapat dicatat atau digunakan untuk analisis lebih lanjut, memungkinkan untuk memantau performa pemrosesan umpan perubahan dan mengidentifikasi potensi hambatan atau penundaan dalam sistem:

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

Contoh delegasi yang menerima perubahan dan menanganinya dengan jeda adalah:

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

Sumber Daya Tambahan:

• Azure Cosmos DB SDK
• Sampel penggunaan di GitHub (.NET)
• Sampel penggunaan di GitHub (Java)
• Sampel tambahan di GitHub

Langkah berikutnya

Sekarang Anda bisa melanjutkan untuk mempelajari selengkapnya tentang prosesor umpan perubahan di artikel berikut ini:

• Ringkasan prosesor umpan perubahan