استخدام موجز التغذية المتغيرة

ينطبق على: NoSQL

توضح هذه المقالة كيف يمكنك رصد تقدم تغيير مثيلات معالج التغذية الخاص بك أثناء قراءة موجز التغيير.

لماذا هو مهم رصد مستوى التقدم؟

يعمل معالج الخلاصة المتغيرة كمؤشر يتحرك للأمام عبرموجز التغيير الخاص بك ويقدم التغييرات على تنفيذ المفوض.

معالج التغذية المتغير يمكن نشره لمعالجة التغييرات بمعدل معين بناءً على الموارد المتاحة مثل وحدة المعالجة المركزية والذاكرة والشبكة وما إلى ذلك.

إذا كان هذا المعدل أبطأ من المعدل الذي تحدث به تغييراتك في حاوية Azure Cosmos DB، يبدأ المعالج في التأخر.

تحديد هذا السيناريو يساعد في فهم ما إذا كنا بحاجة إلى توسيع نطاق نشر معالج التغذية المتغير.

تنفيذ موجز التغيير للتغذية

NET.

كنموذج دفع للإعلامات التلقائية

مثل تغيير معالج التغذية، يمكن أن يعمل موجز التغذية كنموذج دفع. يقيس أداة التقدير الفرق بين آخر عنصر تمت معالجته (تم تعريفه بواسطة حالة حاوية عقود الإيجار) وآخر تغيير في الحاوية، ويدفع هذه القيمة إلى مفوض. يمكن أيضاً تخصيص الفاصل الزمني الذي يكتمل فيه إجراء القياس بقيمة افتراضية تبلغ 5 ثوانٍ.

على سبيل المثال، إذا كان معالج موجز التغيير يستخدم أحدث وضع إصدار وتم تعريفه على النحو التالي:

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

الطريقة الصحيحة لتكوين موجز لقياس هذا المعالج ستستخدم GetChangeFeedEstimatorBuilder مثل:

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

حيث يتشارك كل من المعالج والمقدر في نفس leaseContainer ونفس الاسم.

المعلمتان الأخريان هما المفوض، الذي يتلقى رقما يمثل عدد التغييرات المعلقة التي يجب قراءتها بواسطة المعالج، والفاصل الزمني الذي تريد أخذ هذا القياس عنده.

مثال المفوض الذي يتلقى التقدير هو:

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

يمكنك إرسال هذا التقدير إلى حل المراقبة الخاص بك واستخدامه لفهم كيف تصرف تقدمك بمرور الوقت.

على سبيل التقدير المفصل عند الطلب

على النقيض من نموذج الدفع، هناك بديل يتيح لك الحصول على التقدير عند الطلب. يوفر هذا النموذج أيضاً معلومات مفصلة أكثر:

• التأخرالمقدر لكل عقد إيجار.
• امتلاك المثيل ومعالجة كل عقد إيجار، حتى تتمكن من تحديد ما إن كانت هناك مشكلة على مثيل.

إذا تم تعريف معالج موجز التغيير الخاص بك على النحو التالي:

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

يمكنك إنشاء مقدر مع نفس تكوين عقد التأجير:

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

وكلما أردت ذلك، مع التردد الذي تطلبه، يمكنك الحصول على تقدير مفصل:

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

يحتوي كل ChangeFeedProcessorState منها على معلومات التأجير والتأخر، وكذلك من هو المثيل الحالي الذي يمتلكها.

توزيع المقدر

لا يلزم نشر أداة تقدير موجز التغيير كجزء من معالج موجز التغيير، ولا تكون جزءا من نفس المشروع. نوصي بنشر أداة التقدير على مثيل مستقل عن المعالجات الخاصة بك. يمكن لمثيل مقدر واحد تتبع التقدم لجميع عقود الإيجار والمثيلات في نشر معالج موجز التغيير.

يستهلك كل تقدير وحدات الطلب من حاويات المراقبة والتأجير. يعد التردد 1 دقيقة بينهما نقطة بداية جيدة، كلما انخفض التردد، كلما ارتفعت وحدات الطلب المستهلكة.

Java

يستخدم هذا المثال معالج موجز التغيير في أحدث وضع إصدار مع تقدير التأخير في معالجة أحداث موجز التغيير. في التطبيق - يتم إدراج المستندات في حاوية واحدة ("حاوية الموجز")، وفي الوقت نفسه يقوم مؤشر ترابط عامل آخر أو تطبيق عامل آخر بسحب المستندات المدرجة من موجز تغيير حاوية الموجز والعمل عليها بطريقة ما.

تم إنشاء معالج موجز التغيير وبدأ كما يلي:

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

يمكن إجراء التحقق من تأخر معالج التغذية على تطبيق منفصل (مثل مراقبة السلامة) طالما يتم استخدام نفس حاويات الإدخال (feedContainer و leaseContainer) وبادئة التأجير نفسها بالضبط (CONTAINER_NAME + "-lease"). يتطلب رمز المقدر أن يكون لدى معالج موجز التغيير فرصة لتهيئة مستندات leaseContainer بشكل كامل.

تحسب أداة التقدير التأخير المتراكم عن طريق استرداد الحالة الحالية لمعالج موجز التغيير وتلخيص قيم التأخر المقدرة لكل حدث تتم معالجته:

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

يجب أن يكون إجمالي التأخر في البداية صفرا، وأخيرا يجب أن يكون أكبر أو مساويا لعدد المستندات التي تم إنشاؤها. يمكن تسجيل القيمة الإجمالية للتأخر أو استخدامها لمزيد من التحليل، مما يسمح بمراقبة أداء معالجة موجز التغيير وتحديد أي اختناقات أو تأخيرات محتملة في النظام:

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

مثال على المفوض الذي يتلقى التغييرات ويعالجها بتأخر هو:

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

أوضاع موجز التغيير المدعومة

يمكن استخدام أداة تقدير موجز التغيير لكل من وضع الإصدار الأخير وجميع الإصدارات ووضع الحذف. في كلا الوضعين، لا يضمن التقدير المقدم أن يكون العدد الدقيق للتغييرات المعلقة في العملية.

الموارد الإضافية

• Azure Cosmos DB SDK
• نماذج الاستخدام على GitHub (أحدث إصدار من .NET)
• نماذج الاستخدام على GitHub (.NET جميع الإصدارات والحذف)
• نماذج الاستخدام على GitHub (Java)
• نماذج إضافية على GitHub

الخطوات التالية

يمكنك الآن المتابعة لمعرفة المزيد حول معالج موجز التغيير في المقالة التالية:

• نظرة عامة على معالج موجز التغيير