تغيير التدفقات في واجهة برمجة تطبيقات Azure Cosmos DB لـ MongoDB

ينطبق على: MongoDB

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

إشعار

لاستخدام تدفقات التغيير، أنشئ واجهة برمجة تطبيقات Azure Cosmos DB لحساب MongoDB مع إصدار الخادم 3.6 أو أعلى. إذا قمت بتشغيل أمثلة تدفق التغيير على إصدار سابق، فقد ترى خطأ اسم مرحلة المسار غير المعروف: $ changeStream .

الأمثلة

يوضح المثال التالي كيفية الحصول على تدفقات التغيير على جميع العناصر في المجموعة. ينشئ هذا المثال مؤشرًا لمشاهدة العناصر عند إدراجها أو تحديثها أو استبدالها. مطلوب خيار $match المرحلة، $project والمرحلة fullDocument للحصول على دفقات التغيير. مراقبة عمليات الحذف باستخدام تدفقات التغيير غير مدعوم حاليًا. كحل بديل، يمكنك إضافة علامة ناعمة على العناصر التي يتم حذفها. على سبيل المثال، يمكنك إضافة سمة في العنصر المُسمى "deleted." عندما تريد حذف العنصر، ويمكنك تعيين "deleted" على true وتعيين TTL على العنصر. نظرًا إلى أن تحديث "المحذوفة" إلى true يعد تحديثًا، سيكون هذا التغيير مرئيًا في تدفق التغيير.

JavaScript

var cursor = db.coll.watch(
    [
        { $match: { "operationType": { $in: ["insert", "update", "replace"] } } },
        { $project: { "_id": 1, "fullDocument": 1, "ns": 1, "documentKey": 1 } }
    ],
    { fullDocument: "updateLookup" });

while (!cursor.isExhausted()) {
    if (cursor.hasNext()) {
        printjson(cursor.next());
    }
}

C#‎

var collection = new MongoClient("<connection-string>")
    .GetDatabase("<database-name>")
    .GetCollection<BsonDocument>("<collection-name>");

var pipeline = new EmptyPipelineDefinition<ChangeStreamDocument<BsonDocument>>()
    .Match(change => 
        change.OperationType == ChangeStreamOperationType.Insert || 
        change.OperationType == ChangeStreamOperationType.Update || 
        change.OperationType == ChangeStreamOperationType.Replace
    )
    .AppendStage<ChangeStreamDocument<BsonDocument>, ChangeStreamDocument<BsonDocument>, BsonDocument>(
        @"{ 
            $project: { 
                '_id': 1, 
                'fullDocument': 1, 
                'ns': 1, 
                'documentKey': 1 
            }
        }"
    );

ChangeStreamOptions options = new ()
{
    FullDocument = ChangeStreamFullDocumentOption.UpdateLookup
};

using IChangeStreamCursor<BsonDocument> enumerator = collection.Watch(
    pipeline, 
    options
);

Console.WriteLine("Waiting for changes...");
while (enumerator.MoveNext())
{
    IEnumerable<BsonDocument> changes = enumerator.Current;
    foreach(BsonDocument change in changes)
    {
        Console.WriteLine(change);
    }  
}

Java

يوضح المثال التالي كيفية استخدام وظائف دفق التغيير في Java، للحصول على المثال الكامل، راجع GitHub repo . يوضح هذا المثال أيضًا كيفية استخدام طريقة resumeAfter للبحث عن كافة التغييرات من آخر قراءة.

Bson match = Aggregates.match(Filters.in("operationType", asList("update", "replace", "insert")));

// Pick the field you are most interested in
Bson project = Aggregates.project(fields(include("_id", "ns", "documentKey", "fullDocument")));

// This variable is for second example
BsonDocument resumeToken = null;

// Now time to build the pipeline
List<Bson> pipeline = Arrays.asList(match, project);

//#1 Simple example to seek changes

// Create cursor with update_lookup
MongoChangeStreamCursor<ChangeStreamDocument<org.bson.Document>> cursor = collection.watch(pipeline)
        .fullDocument(FullDocument.UPDATE_LOOKUP).cursor();

Document document = new Document("name", "doc-in-step-1-" + Math.random());
collection.insertOne(document);

while (cursor.hasNext()) {
    // There you go, we got the change document.
    ChangeStreamDocument<Document> csDoc = cursor.next();

    // Let is pick the token which will help us resuming
    // You can save this token in any persistent storage and retrieve it later
    resumeToken = csDoc.getResumeToken();
    //Printing the token
    System.out.println(resumeToken);
    
    //Printing the document.
    System.out.println(csDoc.getFullDocument());
    //This break is intentional but in real project feel free to remove it.
    break;
}

cursor.close();

التغييرات داخل جزء واحد

يوضح المثال التالي كيفية إجراء تغييرات على العناصر داخل جزء واحد. يحصل هذا المثال على تغييرات العناصر التي تحتوي على مفتاح جزء مساوٍ لـ "a" وقيمة مفتاح جزء تساوي "1". من الممكن أن يكون لديك عملاء مختلفون يقرؤون التغييرات من أجزاء مختلفة بالتوازي.

var cursor = db.coll.watch(
    [
        { 
            $match: { 
                $and: [
                    { "fullDocument.a": 1 }, 
                    { "operationType": { $in: ["insert", "update", "replace"] } }
                ]
            }
        },
        { $project: { "_id": 1, "fullDocument": 1, "ns": 1, "documentKey": 1} }
    ],
    { fullDocument: "updateLookup" });

تحجيم تيارات التغيير

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

القيود الحالية

القيود التالية قابلة للتطبيق عند استخدام تدفقات التغيير:

• الخصائص operationType وupdateDescription غير مدعومة بعد في المستند الناتج.
• يتم دعم أنواع العمليات insert وupdate وreplace حاليًا. ومع ذلك، لا يتم دعم عملية الحذف أو الأحداث الأخرى حتى الآن.

نظرًا إلى هذه القيود، يلزم وجود خيارات $ match و$ project stage وfullDocument كما هو موضح في الأمثلة السابقة.

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

معالجة الخطأ

يتم دعم رموز الخطأ والرسائل التالية عند استخدام تدفقات التغيير:

• رمز خطأ HTTP 16500 - عندما يتم تقييد تدفق التغيير، يتم إرجاع صفحة فارغة.

• NamespaceNotFound (OperationType Invalidate) - إذا قمت بتشغيل دفق التغيير على مجموعة غير موجودة أو إذا تم إسقاط المجموعة، فسيتم إرجاع الخطأ NamespaceNotFound. نظرًا إلى أنه لا يمكن إرجاع الخاصية operationType في مستند الإخراج، بدلاً من الخطأ operationType Invalidate، يتم إرجاع الخطأ NamespaceNotFound.

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

• استخدم الوقت للعيش حتى تنتهي صلاحية البيانات تلقائيًا في واجهة برمجة تطبيقات Azure Cosmos DB لـ MongoDB
• الفهرسة في واجهة برمجة تطبيقات Azure Cosmos DB لـ MongoDB