Streamek módosítása az Azure Cosmos DB MongoDB-hez készült API-jában

A KÖVETKEZŐKRE VONATKOZIK: MongoDB

A változáscsatornák támogatása az Azure Cosmos DB MongoDB-hez készült API-jában a change streams API használatával érhető el. A change streams API használatával az alkalmazások lekérhetik a gyűjteményben vagy az egyetlen szegmensben lévő elemeken végrehajtott módosításokat. Később további műveleteket is végrehajthat az eredmények alapján. A gyűjtemény elemeinek módosításait a módosítási idő sorrendjében rögzíti a rendszer, a rendezési sorrend pedig szegmenskulcsonként garantált.

Megjegyzés

A változásstreamek használatához hozza létre a MongoDB-hez készült Azure Cosmos DB API-fiókot a 3.6-os vagy újabb kiszolgálói verzióval. Ha a változásfolyam-példákat egy korábbi verzióval futtatja, a nem felismert folyamatszakasz neve jelenhet meg: $changeStream hiba.

Példák

Az alábbi példa bemutatja, hogyan kérhet le változásstreameket a gyűjtemény összes elemén. Ez a példa egy kurzort hoz létre az elemek beszúrása, frissítése vagy cseréje során történő megtekintéséhez. A $match változásstreamek lekéréséhez szükség van a fázisra, $project a fázisra és fullDocument a beállításra. A változásstreameket használó törlési műveletek figyelése jelenleg nem támogatott. Áthidaló megoldásként hozzáadhat egy helyreállítható jelölőt a törölt elemekhez. Hozzáadhat például egy attribútumot a "törölt" nevű elemhez. Ha törölni szeretné az elemet, beállíthatja a "törölt" true értéket, és beállíthat egy TTL-t az elemen. Mivel a "törölt" true verzió frissítése frissítés, ez a módosítás látható lesz a változásstreamben.

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

Az alábbi példa bemutatja, hogyan használhatja a Stream funkció módosítását a Javában. A teljes példáért tekintse meg ezt a GitHub-adattárat. Ez a példa azt is bemutatja, hogyan használható a metódus az resumeAfter utolsó olvasás összes módosításának keresésére.

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

Változások egyetlen szegmensen belül

Az alábbi példa bemutatja, hogyan kérhet le módosításokat az egyes szegmensek elemein belül. Ez a példa az "a" értékű szegmenskulccsal és az "1"-sel egyenlő szegmenskulcs-értékkel rendelkező elemek módosításait kéri le. Lehetséges, hogy a különböző ügyfelek különböző szegmensekből származó módosításokat olvasnak párhuzamosan.

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

Változásstreamek skálázása

Változásstreamek nagy léptékű használata esetén a legjobb, ha egyenletesen elosztja a terhelést. Használja a GetChangeStreamTokens egyéni parancsot a terhelés fizikai szegmensek/partíciók közötti elosztásához.

Aktuális korlátozások

Változásstreamek használatakor az alábbi korlátozások érvényesek:

• A operationType és updateDescription a tulajdonság még nem támogatott a kimeneti dokumentumban.
• A insert, updateés replace művelettípusok jelenleg támogatottak. A törlési művelet vagy más események azonban még nem támogatottak.

Ezen korlátozások miatt a $match fázisra, $project fázisra és fullDocument beállításokra van szükség az előző példákban látható módon.

Az Azure Cosmos DB NoSQL-hez készült API-jának változáscsatornáival ellentétben nincs külön változáscsatorna-feldolgozó kódtár a változásstreamek felhasználásához, vagy nincs szükség bérlettárolóra. Jelenleg nem támogatott Azure Functions eseményindítók a változásstreamek feldolgozásához.

Hibakezelés

Változásstreamek használatakor az alábbi hibakódok és üzenetek támogatottak:

• 16500-es HTTP-hibakód – Ha a változásstream szabályozva van, üres oldalt ad vissza.

• NamespaceNotFound (OperationType Invalidate) – Ha nem létező módosítási adatfolyamot futtat a gyűjteményen, vagy ha a gyűjtemény el lett dobva, a NamespaceNotFound rendszer hibát ad vissza. Mivel a operationType tulajdonság nem adható vissza a kimeneti dokumentumban a hiba helyett operationType Invalidate , a rendszer a NamespaceNotFound hibát adja vissza.

Következő lépések

• Élettartam használata az adatok automatikus lejáratához az Azure Cosmos DB MongoDB-hez készült API-jában
• Indexelés az Azure Cosmos DB MongoDB-hez készült API-jában