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 daripada tingkat di mana perubahan Anda terjadi di kontainer Azure Cosmos DB Anda, prosesor Anda 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 mengukur perbedaan antara item terakhir yang diproses (ditentukan oleh status 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 menggunakan mode versi terbaru dan 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 menerima angka yang mewakili 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.");
    }
}

Masing-masing ChangeFeedProcessorState berisi informasi sewa dan jeda, dan juga siapa instans saat ini yang memilikinya.

Penyebaran estimator

Estimator umpan perubahan tidak perlu disebarkan sebagai bagian dari prosesor umpan perubahan Anda, juga bukan bagian dari proyek yang sama. Sebaiknya sebarkan estimator pada instans independen dari prosesor Anda. Satu instans estimator dapat melacak kemajuan untuk semua sewa dan instans dalam penyebaran prosesor umpan perubahan Anda.

Setiap estimasi mengonsumsi unit permintaan dari kontainer yang dipantau dan sewa Anda. Frekuensi 1 menit di antaranya adalah titik awal yang baik, semakin rendah frekuensi, semakin tinggi unit permintaan yang digunakan.

Java

Contoh ini menggunakan prosesor umpan perubahan dalam mode versi terbaru dengan estimasi lag 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 dibangun 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 jeda total 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()");
    };
}

Mode umpan perubahan yang didukung

Estimator umpan perubahan dapat digunakan untuk mode versi terbaru dan semua versi dan mode penghapusan. Dalam kedua mode, perkiraan yang disediakan tidak dijamin sebagai jumlah pasti perubahan yang luar biasa untuk diproses.

Sumber Daya Tambahan:

• Azure Cosmos DB SDK
• Sampel penggunaan di GitHub (versi terbaru.NET)
• Sampel penggunaan di GitHub (.NET semua versi dan penghapusan)
• Sampel penggunaan di GitHub (Java)
• Sampel tambahan di GitHub

Langkah berikutnya

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

• Ringkasan prosesor umpan perubahan