Stromen wijzigen in de API van Azure Cosmos DB voor MongoDB

VAN TOEPASSING OP: MongoDB

Ondersteuning voor wijzigingenfeeds in de API van Azure Cosmos DB voor MongoDB is beschikbaar met behulp van de api voor wijzigingsstromen. Met behulp van de api voor wijzigingsstromen kunnen uw toepassingen de wijzigingen in de verzameling of de items in één shard ophalen. Later kunt u verdere acties uitvoeren op basis van de resultaten. Wijzigingen in de items in de verzameling worden vastgelegd in de volgorde van de wijzigingstijd en de sorteervolgorde wordt gegarandeerd per shardsleutel.

Notitie

Als u wijzigingsstromen wilt gebruiken, maakt u de API van Azure Cosmos DB voor het MongoDB-account met serverversie 3.6 of hoger. Als u de voorbeelden van de wijzigingsstroom uitvoert op basis van een eerdere versie, ziet u mogelijk de naam van de niet-herkende pijplijnfase: $changeStream fout.

Voorbeelden

In het volgende voorbeeld ziet u hoe u wijzigingenstromen kunt ophalen voor alle items in de verzameling. In dit voorbeeld wordt een cursor gemaakt om items te bekijken wanneer ze worden ingevoegd, bijgewerkt of vervangen. De $match fase, $project fase en fullDocument optie zijn vereist om de wijzigingsstromen op te halen. Het controleren op verwijderingsbewerkingen met behulp van wijzigingsstromen wordt momenteel niet ondersteund. Als tijdelijke oplossing kunt u een zachte markering toevoegen aan de items die worden verwijderd. U kunt bijvoorbeeld een kenmerk toevoegen in het item met de naam 'verwijderd'. Wanneer u het item wilt verwijderen, kunt u 'verwijderd' true instellen op en een TTL instellen voor het item. Omdat het bijwerken van 'verwijderd' naar true een update is, is deze wijziging zichtbaar in de wijzigingsstroom.

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

In het volgende voorbeeld ziet u hoe u de functionaliteit van de wijzigingsstroom in Java gebruikt, voor het volledige voorbeeld, deze GitHub-opslagplaats. In dit voorbeeld ziet u ook hoe u de resumeAfter methode gebruikt om alle wijzigingen van de laatste leesbewerking te zoeken.

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

Wijzigingen binnen één shard

In het volgende voorbeeld ziet u hoe u wijzigingen kunt aanbrengen in de items binnen één shard. In dit voorbeeld worden de wijzigingen opgehaald van items met een shardsleutel die gelijk is aan 'a' en de shardsleutelwaarde gelijk aan '1'. Het is mogelijk dat verschillende clients wijzigingen van verschillende shards parallel lezen.

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

Wijzigingsstromen schalen

Wanneer u wijzigingsstromen op schaal gebruikt, kunt u de belasting het beste gelijkmatig verdelen. Gebruik de aangepaste opdracht GetChangeStreamTokens om de belasting over fysieke shards/partities te verdelen.

Huidige beperkingen

De volgende beperkingen zijn van toepassing bij het gebruik van wijzigingsstromen:

• De operationType en updateDescription eigenschappen worden nog niet ondersteund in het uitvoerdocument.
• De inserttypen , updateen replace bewerkingen worden momenteel ondersteund. De verwijderbewerking of andere gebeurtenissen worden echter nog niet ondersteund.

Vanwege deze beperkingen zijn de $match fase, $project fase en fullDocument-opties vereist, zoals wordt weergegeven in de vorige voorbeelden.

In tegenstelling tot de wijzigingenfeed in de API van Azure Cosmos DB voor NoSQL, is er geen afzonderlijke bibliotheek met wijzigingenfeedprocessors voor het verbruik van wijzigingsstromen of een behoefte aan een leasecontainer. Er is momenteel geen ondersteuning voor Azure Functions-triggers om wijzigingsstromen te verwerken.

Foutafhandeling

De volgende foutcodes en berichten worden ondersteund bij het gebruik van wijzigingsstromen:

• HTTP-foutcode 16500 : wanneer de wijzigingsstroom wordt beperkt, wordt er een lege pagina geretourneerd.

• NamespaceNotFound (OperationType Invalidate): als u een wijzigingsstroom uitvoert op de verzameling die niet bestaat of als de verzameling wordt verwijderd, wordt er een NamespaceNotFound fout geretourneerd. Omdat de operationType eigenschap niet kan worden geretourneerd in het uitvoerdocument, in plaats van de operationType Invalidate fout, wordt de NamespaceNotFound fout geretourneerd.

Volgende stappen

• Time to Live gebruiken om gegevens automatisch te laten verlopen in de API van Azure Cosmos DB voor MongoDB
• Indexeren in de API van Azure Cosmos DB voor MongoDB