قم بتغيير معالج موجز التغيير في Azure Cosmos DB

معالج موجز التغيير هو جزء من Azure Cosmos DB .NET V3وJava V4 SDKs، والتي تبسط عملية قراءة موجز التغيير وتوزيع معالجة الحدث عبر العديد من المستهلكين بشكل فعال.

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

عدة تطوير البرامج SDK مدعومة

.NET V3	Java	Node.JS	Python
✓	✓	✕	✕

مكونات معالج موجز التغيير

يحتوي معالج موجز التغيير على أربعة مكونات رئيسية:

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

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

• مثيل الحساب : يستضيف مثيل الحساب معالج خلاصة التغيير للاستماع إلى التغييرات. اعتمادا على النظام الأساسي، قد يتم تمثيله بواسطة جهاز ظاهري (VM) أو جراب Kubernetes أو مثيل Azure App Service أو جهاز فعلي. يحتوي مثيل الحساب على معرف فريد يسمى اسم المثيل خلال هذه المقالة.

• المفوض: المفوض هو التعليمات البرمجية التي تحدد ما تريد أنت، المطور، القيام به مع كل دفعة من التغييرات التي يقرأها معالج موجز التغيير.

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

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

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

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

.NET

يتوفر معالج موجز التغيير في .NET لأحدث وضع إصدار وجميع الإصدارات ووضع الحذف. جميع الإصدارات ووضع الحذف قيد المعاينة ومدعم لمعالج موجز التغيير الذي يبدأ في الإصدار 3.40.0-preview.0. نقطة الدخول لكلا الوضعين هي دائما الحاوية المراقبة.

للقراءة باستخدام أحدث وضع إصدار، في مثيل Container ، يمكنك استدعاء GetChangeFeedProcessorBuilder:

/// <summary>
/// Start the Change Feed Processor to listen for changes and process them with the HandleChangesAsync implementation.
/// </summary>
private static async Task<ChangeFeedProcessor> StartChangeFeedProcessorAsync(
    CosmosClient cosmosClient,
    IConfiguration configuration)
{
    string databaseName = configuration["SourceDatabaseName"];
    string sourceContainerName = configuration["SourceContainerName"];
    string leaseContainerName = configuration["LeasesContainerName"];

    Container leaseContainer = cosmosClient.GetContainer(databaseName, leaseContainerName);
    ChangeFeedProcessor changeFeedProcessor = cosmosClient.GetContainer(databaseName, sourceContainerName)
        .GetChangeFeedProcessorBuilder<ToDoItem>(processorName: "changeFeedSample", onChangesDelegate: HandleChangesAsync)
            .WithInstanceName("consoleHost")
            .WithLeaseContainer(leaseContainer)
            .Build();

    Console.WriteLine("Starting Change Feed Processor...");
    await changeFeedProcessor.StartAsync();
    Console.WriteLine("Change Feed Processor started.");
    return changeFeedProcessor;
}

للقراءة باستخدام جميع الإصدارات ووضع الحذف، اتصل GetChangeFeedProcessorBuilderWithAllVersionsAndDeletes من المثيل Container :

Container leaseContainer = client.GetContainer(Program.databaseName, Program.leasesContainer);
Container monitoredContainer = client.GetContainer(Program.databaseName, containerName);
ChangeFeedProcessor changeFeedProcessor = monitoredContainer
    .GetChangeFeedProcessorBuilderWithAllVersionsAndDeletes<ToDoItem>(processorName: "changeFeedBasic", onChangesDelegate: Program.HandleChangesAsync)
        .WithInstanceName("consoleHost")
        .WithLeaseContainer(leaseContainer)
        .Build();

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

فيما يلي مثال على مفوض لأحدث وضع إصدار:

/// <summary>
/// The delegate receives batches of changes as they are generated in the change feed and can process them.
/// </summary>
static async Task HandleChangesAsync(
    ChangeFeedProcessorContext context,
    IReadOnlyCollection<ToDoItem> changes,
    CancellationToken cancellationToken)
{
    Console.WriteLine($"Started handling changes for lease {context.LeaseToken}...");
    Console.WriteLine($"Change Feed request consumed {context.Headers.RequestCharge} RU.");
    // SessionToken if needed to enforce Session consistency on another client instance
    Console.WriteLine($"SessionToken ${context.Headers.Session}");

    // We may want to track any operation's Diagnostics that took longer than some threshold
    if (context.Diagnostics.GetClientElapsedTime() > TimeSpan.FromSeconds(1))
    {
        Console.WriteLine($"Change Feed request took longer than expected. Diagnostics:" + context.Diagnostics.ToString());
    }

    foreach (ToDoItem item in changes)
    {
        Console.WriteLine($"Detected operation for item with id {item.id}, created at {item.creationTime}.");
        // Simulate some asynchronous operation
        await Task.Delay(10);
    }

    Console.WriteLine("Finished handling changes.");
}

فيما يلي مثال على مفوض لكافة الإصدارات ووضع الحذف:

static async Task HandleChangesAsync(ChangeFeedProcessorContext context, IReadOnlyCollection<ChangeFeedItem<ToDoItem>> changes, CancellationToken cancellationToken)
{
    Console.WriteLine($"Started handling changes for lease {context.LeaseToken}...");
    Console.WriteLine($"Change Feed request consumed {context.Headers.RequestCharge} RU.");
    // SessionToken if needed to enforce Session consistency on another client instance
    Console.WriteLine($"SessionToken ${context.Headers.Session}");

    // We may want to track any operation's Diagnostics that took longer than some threshold
    if (context.Diagnostics.GetClientElapsedTime() > TimeSpan.FromSeconds(1))
    {
        Console.WriteLine($"Change Feed request took longer than expected. Diagnostics:" + context.Diagnostics.ToString());
    }

    foreach (ChangeFeedItem<ToDoItem> item in changes)
    {
        if (item.Metadata.OperationType == ChangeFeedOperationType.Delete)
        {
            Console.WriteLine($"\tDetected {item.Metadata.OperationType} operation for item.");
        }
        else
        {
            Console.WriteLine($"\tDetected {item.Metadata.OperationType} operation for item with id {item.Current.id}.");
        }
        // Simulate work
        await Task.Delay(1);
    }
}

بعد ذلك، يمكنك تعريف اسم مثيل الحساب أو المعرف الفريد باستخدام WithInstanceName. يجب أن يكون اسم مثيل الحساب فريدا ومختلفا لكل مثيل حساب تقوم بنشره. يمكنك تعيين الحاوية للحفاظ على حالة التأجير باستخدام WithLeaseContainer.

يمنحك الاتصال Build مثيل المعالج الذي يمكنك البدء به عن طريق استدعاء StartAsync.

مهم

عند إنشاء كل CosmosClient من حاويات التغذية والإيجار، وتهيئة عبء عمل معالج التغيير الجديد في الخلاصة:

استخدم نقطة نهاية عامة

• حدد دائما نقطة نهاية عالمية (على سبيل المثال، contoso.documents.azure.com) بدلا من نقطة نهاية إقليمية (على سبيل المثال، contoso-westus.documents.azure.com).

تبديل المناطق باستخدام ApplicationRegion أو ApplicationPreferredRegions

• لإعادة توجيه حركة البيانات بين المناطق، اعتمد على ApplicationRegion خاصية OR ApplicationPreferredRegions .
• يقوم معالج تغيير التغذية بإنشاء مستندات إيجار مرتبطة بنقطة النهاية المكونة، وبالتالي يؤدي تغيير نقاط النهاية إلى إنشاء مستندات إيجار مستقلة جديدة.

✅ قم بذلك - استخدم نقطة النهاية العالمية مع ApplicationRegion:

CosmosClient client = new CosmosClient(
    "https://contoso.documents.azure.com:443/",  // Global endpoint
    "<account-key>",
    new CosmosClientOptions()
    {
        ApplicationRegion = Regions.WestUS2  // Specify region here
    });

Container monitoredContainer = client.GetContainer("myDatabase", "myContainer");
Container leaseContainer = client.GetContainer("myDatabase", "leases");

✅ قم بذلك - استخدم نقطة النهاية العالمية مع ApplicationPreferredRegions:

CosmosClient client = new CosmosClient(
    "https://contoso.documents.azure.com:443/",  // Global endpoint
    "<account-key>",
    new CosmosClientOptions()
    {
        ApplicationPreferredRegions = new List<string> { Regions.WestUS2, Regions.EastUS2 }
    });

Container monitoredContainer = client.GetContainer("myDatabase", "myContainer");
Container leaseContainer = client.GetContainer("myDatabase", "leases");

❌ لا تفعل ذلك - تجنب نقاط النهاية الإقليمية:

// DON'T: Using regional endpoint will create region-scoped lease documents
CosmosClient client = new CosmosClient(
    "https://contoso-westus.documents.azure.com:443/",  // Regional endpoint - AVOID
    "<account-key>");

مهم

تجنب المعالجة غير المتزامنة في أساليب المفوض: عند استخدام واجهات برمجة التطبيقات غير المتزامنة داخل أسلوب المفوض، handleChanges() كن على دراية بأن معالج موجز التغيير قد يتحقق من عقد الإيجار قبل اكتمال جميع العمليات غير المتزامنة. يمكن أن يؤدي ذلك إلى أحداث فائتة إذا واجه التطبيق مشكلات أثناء الاسترداد. ضع في اعتبارك استخدام المعالجة المتزامنة أو تنفيذ تتبع الإكمال المناسب قبل السماح للمفوض بالعودة.

إشعار

يتم أخذ قصاصات التعليمات البرمجية السابقة من عينات في GitHub. يمكنك الحصول على نموذج لأحدث وضع إصدار أو جميع الإصدارات ووضع الحذف.

دورة حياة المعالجة

دورة الحياة العادية لمثيل المضيف هي:

1. اقرأ موجز التغيير.
2. إذا لم تكن هناك تغييرات، فانتقل إلى الخطوة 1 لمدة محددة مسبقا (قابلة للتخصيص باستخدام WithPollInterval في Builder).
3. إذا كانت هناك تغييرات، أرسلها إلى المفوض.
4. عندما ينتهي المفوض من معالجة التغييرات بنجاح، قم بتحديث مخزن التأجير بأحدث نقطة معالجة في الوقت المناسب وانتقل إلى الخطوة 1.

معالجة الأخطاء

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

إشعار

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

لمنع معالج موجز التغيير من التعثر باستمرار في إعادة محاولة دفعة التغييرات نفسها، يجب إضافة منطق في التعليمات البرمجية للمفوض لكتابة المستندات، عند الاستثناء، إلى قائمة انتظار رسالة خطأ. يضمن هذا التصميم أنه يمكنك تتبع التغييرات غير المعالجة بينما تظل قادراً على الاستمرار في معالجة التغييرات المستقبلية. قد تكون قائمة انتظار الرسائل الخطأ حاوية Azure Cosmos DB أخرى. مخزن البيانات الدقيق لا يهم. تريد ببساطة استمرار التغييرات غير المعالجة.

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

تأكد من أن مهلة طلب شبكة العميل أطول من المهلة من جانب الخادم لمنع عدم تطابق المهلة التي يمكن أن تؤدي إلى توقف المعالجة. المهلة الافتراضية من جانب الخادم هي 5 ثوان. راقب الأخطاء المتعلقة بالمهلة واضبط التكوينات وفقا لذلك.

إعلامات دورة الحياة

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

• قم بتسجيل معالج WithLeaseAcquireNotification ليتم إخطاره عندما يحصل المضيف الحالي على عقد إيجار لبدء معالجته.
• قم بتسجيل معالج WithLeaseReleaseNotification ليتم إخطاره عندما يصدر المضيف الحالي عقد إيجار ويتوقف عن معالجته.
• تسجيل معالج WithErrorNotification ليتم إعلامه عندما يواجه المضيف الحالي استثناء أثناء المعالجة. يجب أن تكون قادرا على تمييز ما إذا كان المصدر هو مفوض المستخدم (استثناء غير معالج) أو خطأ يواجهه المعالج عندما يحاول الوصول إلى الحاوية المراقبة (على سبيل المثال، مشكلات الشبكة).

تتوفر إعلامات دورة الحياة في وضعي موجز التغيير. فيما يلي مثال على إعلامات دورة الحياة في أحدث وضع للإصدار:

Container.ChangeFeedMonitorLeaseAcquireDelegate onLeaseAcquiredAsync = (string leaseToken) =>
{
    Console.WriteLine($"Lease {leaseToken} is acquired and will start processing");
    return Task.CompletedTask;
};

Container.ChangeFeedMonitorLeaseReleaseDelegate onLeaseReleaseAsync = (string leaseToken) =>
{
    Console.WriteLine($"Lease {leaseToken} is released and processing is stopped");
    return Task.CompletedTask;
};

Container.ChangeFeedMonitorErrorDelegate onErrorAsync = (string LeaseToken, Exception exception) =>
{
    if (exception is ChangeFeedProcessorUserException userException)
    {
        Console.WriteLine($"Lease {LeaseToken} processing failed with unhandled exception from user delegate {userException.InnerException}");
    }
    else
    {
        Console.WriteLine($"Lease {LeaseToken} failed with {exception}");
    }

    return Task.CompletedTask;
};

ChangeFeedProcessor changeFeedProcessor = monitoredContainer
    .GetChangeFeedProcessorBuilder<ToDoItem>("changeFeedNotifications", handleChanges)
        .WithLeaseAcquireNotification(onLeaseAcquiredAsync)
        .WithLeaseReleaseNotification(onLeaseReleaseAsync)
        .WithErrorNotification(onErrorAsync)
        .WithInstanceName("consoleHost")
        .WithLeaseContainer(leaseContainer)
        .Build();

وحدة الانتشار

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

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

تحجيم ديناميكي

كما ذكرنا سابقا، داخل وحدة التوزيع، يمكن أن يكون لديك مثيل حساب واحد أو أكثر. للاستفادة من توزيع الحوسبة داخل وحدة التوزيع، المتطلبات الرئيسية الوحيدة هي:

• يجب أن تحتوي جميع المثيلات على نفس تكوين حاوية الإيجار.
• يجب أن يكون لكافة المثيلات نفس القيمة ل processorName.
• يجب أن يكون لكل مثيل اسم مثيل مختلف (WithInstanceName).

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

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

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

وقت البدء

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

القراءة من تاريخ ووقت سابقين

من الممكن تهيئة معالج موجز التغيير لقراءة التغييرات بدءا من تاريخ ووقت محددين عن طريق تمرير مثيل DateTime إلى ملحق المنشئ WithStartTime :

Container leaseContainer = client.GetContainer(databaseId, Program.leasesContainer);
Container monitoredContainer = client.GetContainer(databaseId, Program.monitoredContainer);
ChangeFeedProcessor changeFeedProcessor = monitoredContainer
    .GetChangeFeedProcessorBuilder<ToDoItem>("changeFeedTime", Program.HandleChangesAsync)
        .WithInstanceName("consoleHost")
        .WithLeaseContainer(leaseContainer)
        .WithStartTime(particularPointInTime)
        .Build();

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

القراءة من البداية

في سيناريوهات أخرى، كما هو الحال في عمليات ترحيل البيانات أو إذا كنت تقوم بتحليل محفوظات الحاوية بأكملها، فأنت بحاجة إلى قراءة موجز التغيير من بداية عمر تلك الحاوية. يمكنك استخدام WithStartTime على ملحق المنشئ، ولكن تمرير DateTime.MinValue.ToUniversalTime()، الذي ينشئ تمثيل UTC للحد الأدنى DateTime للقيمة كما في هذا المثال:

Container leaseContainer = client.GetContainer(databaseId, Program.leasesContainer);
Container monitoredContainer = client.GetContainer(databaseId, Program.monitoredContainer);
ChangeFeedProcessor changeFeedProcessor = monitoredContainer
    .GetChangeFeedProcessorBuilder<ToDoItem>("changeFeedBeginning", Program.HandleChangesAsync)
        .WithInstanceName("consoleHost")
        .WithLeaseContainer(leaseContainer)
        .WithStartTime(DateTime.MinValue.ToUniversalTime())
        .Build();

تتم تهيئة معالج موجز التغيير، ويبدأ في قراءة التغييرات من بداية عمر الحاوية.

إشعار

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

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

Java

للحصول على عينات العمل الكاملة، راجع مستودع GitHub هذا. مثال على تنفيذ المفوض عند قراءة موجز التغيير في أحدث وضع للإصدار هو:

private static Consumer<List<JsonNode>> handleChanges() {
    return (List<JsonNode> docs) -> {
        logger.info("Start handleChanges()");

        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();
            }
        }
        isWorkCompleted = true;
        logger.info("End handleChanges()");

    };
}

إشعار

في هذا المثال، يمكنك تمرير متغير options من النوع ChangeFeedProcessorOptions، والذي يمكن استخدامه لتعيين قيم مختلفة، بما في ذلك setStartFromBeginning:

options = new ChangeFeedProcessorOptions();
options.setStartFromBeginning(false);
options.setLeasePrefix("myChangeFeedDeploymentUnit");
options.setFeedPollDelay(Duration.ofSeconds(5));
options.setFeedPollThroughputControlConfig(throughputControlGroupConfig);

تطبيق المفوض لقراءة موجز التغيير في جميع الإصدارات ووضع الحذف مشابه، ولكن بدلا من استدعاء .handleChanges()، قم باستدعاء .handleAllVersionsAndDeletesChanges(). وضع جميع الإصدارات والحذف قيد المعاينة ومتوفر في إصدار >Java SDK = 4.42.0.

فيما يلي مثال:

public static ChangeFeedProcessor getChangeFeedProcessorForAllVersionsAndDeletesMode(String hostName, CosmosAsyncContainer feedContainer, CosmosAsyncContainer leaseContainer) {
    return new ChangeFeedProcessorBuilder()
            .hostName(hostName)
            .options(options)
            .feedContainer(feedContainer)
            .leaseContainer(leaseContainer)
            .handleAllVersionsAndDeletesChanges((List<ChangeFeedProcessorItem> changeFeedProcessorItems) -> {
                logger.info("--->handleAllVersionsAndDeletesChanges() START");

                for (ChangeFeedProcessorItem item : changeFeedProcessorItems) {
                    try {
                        // AllVersionsAndDeletes Change Feed hands the document to you in the form of ChangeFeedProcessorItem
                        // As a developer you have two options for handling the ChangeFeedProcessorItem provided to you by Change Feed
                        // One option is to operate on the item as it is and call the different getters for different states, as shown below.
                        logger.info("---->DOCUMENT RECEIVED: {}", OBJECT_MAPPER.writerWithDefaultPrettyPrinter()
                                .writeValueAsString(item));

                        logger.info("---->CURRENT RECEIVED: {}", item.getCurrent());
                        logger.info("---->PREVIOUS RECEIVED: {}", item.getPrevious());
                        logger.info("---->METADATA RECEIVED: {}", item.getChangeFeedMetaData());

                        // You can also transform the ChangeFeedProcessorItem to JsonNode and work on the generic json structure.
                        // This is great especially if you do not have a single uniform data model for all documents.
                        logger.info("----=>JsonNode received: " + item.toJsonNode());

                    } catch (JsonProcessingException e) {
                        e.printStackTrace();
                    }
                }
                logger.info("--->handleAllVersionsAndDeletesChanges() END");

            })
            .buildChangeFeedProcessor();
}

في أي من وضع موجز التغيير، يمكنك تعيينه إلى changeFeedProcessorInstance وتمرير معلمات اسم مثيل الحساب (hostName)، والحاوية المراقبة (تسمى feedContainerهنا )، و leaseContainer. ثم ابدأ معالج موجز التغيير:

logger.info("Start Change Feed Processor on worker (handles changes asynchronously)");
ChangeFeedProcessor changeFeedProcessorInstance = new ChangeFeedProcessorBuilder()
    .hostName("SampleHost_1")
    .feedContainer(feedContainer)
    .leaseContainer(leaseContainer)
    .handleChanges(handleChanges())
    .options(options)
    .buildChangeFeedProcessor();
changeFeedProcessorInstance.start()
                           .subscribeOn(Schedulers.boundedElastic())
                           .subscribe();

مهم

عند إنشاء كل CosmosAsyncClient من حاويات التغذية والإيجار، وتهيئة عبء عمل معالج التغيير الجديد في الخلاصة:

استخدم نقطة نهاية عامة

• حدد دائما نقطة نهاية عالمية (على سبيل المثال، contoso.documents.azure.com) بدلا من نقطة نهاية إقليمية (على سبيل المثال، contoso-westus.documents.azure.com).

تبديل المناطق باستخدام المناطق المفضلة

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

✅ قم بذلك - استخدم نقطة النهاية العالمية مع المناطق المفضلة:

List<String> preferredRegions = new ArrayList<>();
preferredRegions.add("West US 2");
preferredRegions.add("East US 2");

CosmosAsyncClient client = new CosmosClientBuilder()
    .endpoint("https://contoso.documents.azure.com:443/")  // Global endpoint
    .key("<account-key>")
    .preferredRegions(preferredRegions)  // Specify regions here
    .buildAsyncClient();

CosmosAsyncContainer feedContainer = client
    .getDatabase("myDatabase")
    .getContainer("myContainer");
CosmosAsyncContainer leaseContainer = client
    .getDatabase("myDatabase")
    .getContainer("leases");

❌ لا تفعل ذلك - تجنب نقاط النهاية الإقليمية:

// DON'T: Using regional endpoint will create region-scoped lease documents
CosmosAsyncClient client = new CosmosClientBuilder()
    .endpoint("https://contoso-westus.documents.azure.com:443/")  // Regional endpoint - AVOID
    .key("<account-key>")
    .buildAsyncClient();

مهم

تجنب المعالجة غير المتزامنة في أساليب المفوض: عند استخدام واجهات برمجة التطبيقات غير المتزامنة داخل أسلوب المفوض، handleChanges() كن على دراية بأن معالج موجز التغيير قد يتحقق من عقد الإيجار قبل اكتمال جميع العمليات غير المتزامنة. يمكن أن يؤدي ذلك إلى أحداث فائتة إذا واجه التطبيق مشكلات أثناء الاسترداد. ضع في اعتبارك استخدام المعالجة المتزامنة أو تنفيذ تتبع الإكمال المناسب قبل السماح للمفوض بالعودة.

إشعار

يتم أخذ قصاصات التعليمات البرمجية السابقة من عينات في GitHub. يمكنك الحصول على نموذج لأحدث وضع إصدار أو جميع الإصدارات ووضع الحذف.

دورة حياة المعالجة

دورة الحياة العادية لمثيل المضيف هي:

1. اقرأ موجز التغيير.
2. إذا لم تكن هناك تغييرات، فانتقل إلى الخطوة 1 لفترة محددة مسبقا (قابلة للتخصيص مع options.setFeedPollDelay في المنشئ).
3. إذا كانت هناك تغييرات، أرسلها إلى المفوض.
4. عندما ينتهي المفوض من معالجة التغييرات بنجاح، قم بتحديث مخزن التأجير باستخدام أحدث نقطة معالجة في الوقت المناسب وانتقل إلى الخطوة 1.

معالجة الأخطاء

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

إشعار

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

لمنع معالج موجز التغيير من "التعثر" باستمرار في إعادة محاولة دفعة التغييرات نفسها، يجب إضافة منطق في التعليمات البرمجية للمفوض لكتابة المستندات، عند الاستثناء، إلى رسالة خطأ. يضمن هذا التصميم أنه يمكنك تتبع التغييرات غير المعالجة بينما تظل قادراً على الاستمرار في معالجة التغييرات المستقبلية. قد تكون الرسالة التي تم خطأها حاوية Azure Cosmos DB أخرى. مخزن البيانات الدقيق لا يهم. تريد ببساطة استمرار التغييرات غير المعالجة.

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

تأكد من أن مهلة طلب شبكة العميل أطول من المهلة من جانب الخادم لمنع عدم تطابق المهلة التي يمكن أن تؤدي إلى توقف المعالجة. المهلة الافتراضية من جانب الخادم هي 5 ثوان. راقب الأخطاء المتعلقة بالمهلة واضبط التكوينات وفقا لذلك.

وحدة الانتشار

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

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

تحجيم ديناميكي

كما ذكرنا سابقا، داخل وحدة التوزيع، يمكن أن يكون لديك مثيل حساب واحد أو أكثر. للاستفادة من توزيع الحوسبة داخل وحدة التوزيع، المتطلبات الرئيسية الوحيدة هي:

• يجب أن يكون لكافة المثيلات نفس تكوين حاوية التأجير
• يجب أن يكون لكافة المثيلات نفس القيمة المعينة في options.setLeasePrefix (أو لم يتم تعيين أي منها على الإطلاق)
• يجب أن يكون لكل مثيل مثيل مختلف hostName

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

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

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

وقت البدء

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

إشعار

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

القراءة من تاريخ ووقت سابقين

من الممكن تهيئة معالج موجز التغيير لقراءة التغييرات بدءا من تاريخ ووقت محددين عن طريق الإعداد setStartTime في options. تتم تهيئة معالج موجز التغيير لهذا التاريخ والوقت المحددين، ويبدأ في قراءة التغييرات التي حدثت بعد ذلك.

القراءة من البداية

في النموذج، setStartFromBeginning يتم تعيين إلى false، وهو نفس القيمة الافتراضية. في سيناريوهات أخرى، كما هو الحال في عمليات ترحيل البيانات أو إذا كنت تقوم بتحليل محفوظات الحاوية بأكملها، فأنت بحاجة إلى قراءة موجز التغيير من بداية عمر تلك الحاوية. للقيام بذلك، يمكنك تعيين setStartFromBeginning إلى true. تتم تهيئة معالج موجز التغيير، ويبدأ في قراءة التغييرات من بداية عمر الحاوية.

إشعار

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

موجز التغيير والإنتاجية المقدمة

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

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

مشاركة حاوية التأجير

يمكنك مشاركة حاوية تأجير عبر وحدات توزيع متعددة. في حاوية تأجير مشتركة، تستمع كل وحدة توزيع إلى حاوية مراقبة مختلفة أو لها قيمة مختلفة ل processorName. في هذا التكوين، تحتفظ كل وحدة توزيع بحالة مستقلة على حاوية التأجير. راجع استهلاك وحدة الطلب على حاوية تأجير للتأكد من أن معدل النقل المقدم كاف لجميع وحدات التوزيع.

تكوين التأجير المتقدم

يمكن أن تؤثر ثلاثة تكوينات رئيسية على كيفية عمل معالج موجز التغيير. يؤثر كل تكوين على استهلاك وحدة الطلب على حاوية التأجير. يمكنك تعيين أحد هذه التكوينات عند إنشاء معالج موجز التغيير، ولكن استخدمها بعناية:

• التأجير الحصول: بشكل افتراضي، كل 17 ثانية. يتحقق المضيف بشكل دوري من حالة مخزن التأجير ويفكر في الحصول على عقود الإيجار كجزء من عملية التحجيم الديناميكي. تتم هذه العملية عن طريق تنفيذ استعلام على حاوية التأجير. يؤدي تقليل هذه القيمة إلى جعل إعادة التوازن والحصول على عقود الإيجار أسرع، ولكنه يزيد من استهلاك وحدة الطلب على حاوية التأجير.
• انتهاء صلاحية التأجير: بشكل افتراضي، 60 ثانية. يحدد الحد الأقصى لمقدار الوقت الذي يمكن أن يتواجد فيه عقد الإيجار دون أي نشاط تجديد قبل الحصول عليه من قبل مضيف آخر. عند تعطل مضيف، يتم التقاط عقود الإيجار التي يمتلكها من قبل مضيفين آخرين بعد هذه الفترة الزمنية بالإضافة إلى الفاصل الزمني للتجديد المكون. يؤدي تقليل هذه القيمة إلى إجراء الاسترداد بعد تعطل المضيف بشكل أسرع، ولكن يجب ألا تكون قيمة انتهاء الصلاحية أقل من الفاصل الزمني للتجديد.
• تجديد التأجير: بشكل افتراضي، كل 13 ثانية. يقوم المضيف الذي يمتلك عقد إيجار بتجديد عقد الإيجار بشكل دوري، حتى إذا لم تكن هناك تغييرات جديدة للاستهلاك. تتم هذه العملية عن طريق تنفيذ استبدال في عقد الإيجار. يؤدي تقليل هذه القيمة إلى تقليل الوقت المطلوب للكشف عن عقود الإيجار المفقودة بسبب تعطل المضيف، ولكنه يزيد من استهلاك وحدة الطلب على حاوية التأجير.

أين تستضيف معالج موجز التغيير

يمكن استضافة معالج موجز التغيير في أي نظام أساسي يدعم العمليات أو المهام طويلة الأمد. إليك بعض الأمثلة:

• عملية في مثيل أجهزة Azure الظاهرية
• وظيفة خلفية في خدمة Azure Kubernetes
• دالة بلا خادم في Azure Functions
• خدمة مستضافة ASP.NET

على الرغم من أن معالج موجز التغيير يمكن تشغيله في بيئات قصيرة الأجل لأن حاوية التأجير تحافظ على الحالة، فإن دورة بدء التشغيل لهذه البيئات تضيف تأخيرات إلى الوقت الذي يستغرقه تلقي الإعلامات (بسبب الحمل الزائد لبدء تشغيل المعالج في كل مرة يتم فيها بدء تشغيل البيئة).

متطلبات الوصول المستندة إلى الدور

عند استخدام معرف Microsoft Entra كآلية مصادقة، تأكد من أن الهوية لديها الأذونات المناسبة:

• في الحاوية المراقبة:
  • Microsoft.DocumentDB/databaseAccounts/readMetadata
  • Microsoft.DocumentDB/databaseAccounts/sqlDatabases/containers/readChangeFeed
• في حاوية التأجير:
  • Microsoft.DocumentDB/databaseAccounts/sqlDatabases/containers/items/read
  • Microsoft.DocumentDB/databaseAccounts/sqlDatabases/containers/items/create
  • Microsoft.DocumentDB/databaseAccounts/sqlDatabases/containers/items/replace
  • Microsoft.DocumentDB/databaseAccounts/sqlDatabases/containers/items/delete
  • Microsoft.DocumentDB/databaseAccounts/sqlDatabases/containers/items/executeQuery

موارد إضافية

• Azure Cosmos DB SDK
• إكمال نموذج التطبيق على GitHub
• نماذج الاستخدام على GitHub
• مختبرات ورشة عمل Azure Cosmos DB لمعالج موجز التغيير

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

تعرف على المزيد حول معالج موجز التغيير في المقالات التالية:

• نظرة عامة على موجز التغيير
• تغيير نموذج سحب الموجز
• كيفية الترحيل من مكتبة معالج موجز التغيير
• استخدام أداة تقدير موجز التغيير
• تغيير وقت بدء معالج الموجز