Ändra Stream på vCore-baserade Azure Cosmos DB för MongoDB (förhandsversion)

Ändringsströmmar är en realtidsström med databasändringar som flödar från databasen till ditt program. Med den här funktionen kan du skapa reaktiva program genom att prenumerera på databasändringar, vilket eliminerar behovet av kontinuerlig avsökning för att identifiera ändringar.

Aktivera ändringsströmmar

Du kan aktivera eller inaktivera den här funktionen med hjälp av Azure CLI eller en ARM-mall. Portalstöd kommer snart att läggas till.

Steg för att aktivera ändringsströmmar i vCore-kluster via CLI

1. Logga in på Azure CLI

az login

2. Hämta de aktuella inställningarna för funktionsflaggor i klustret. Detta säkerställer att du behåller alla befintliga flaggor när du lägger till den nya funktionen.

az resource show --ids "/subscriptions/<sub id>/resourceGroups/<resource group name>/providers/Microsoft.DocumentDB/mongoClusters/<resource name of your Cosmos DB for MongoDB vCore cluster>" --api-version 2024-10-01-preview

3. Skicka PATCH-begäran för att aktivera funktionen.

az resource patch --ids "/subscriptions/<subscription_id>/resourceGroups/<resource_group_name>/providers/Microsoft.DocumentDB/mongoClusters/<vCore_cluster_name>" --api-version 2024-10-01-preview --properties "{\"previewFeatures\": [ \"ChangeStreams\"]}"

4. Kontrollera resultatet:
   • Se till att svarsnyttolasten innehåller "previewFeatures": ["ChangeStreams"].
   • Om du får felet "ändringsströmmar stöds inte i det här klustret" skapar du en supportbegäran.

Konfigurera ändringsströmmar

Den här exempelkoden initierar en ändringsström i exampleCollection samlingen och övervakar kontinuerligt eventuella ändringar. När en ändring identifieras hämtar den ändringshändelsen och skriver ut den i JSON-format.

JavaScript

// Open a change stream
const changeStream = db.exampleCollection.watch();

// Listen for changes
while (changeStream.hasNext()) 
    {
        const change = changeStream.next();
        printjson(change);
    }

Java

package com.example;
 
import com.mongodb.MongoClientSettings;
import com.mongodb.client.MongoClients;
import com.mongodb.client.MongoClient;
import com.mongodb.client.MongoDatabase;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoCursor;
import com.mongodb.client.model.changestream.ChangeStreamDocument;
 
import org.bson.BsonDocument;
import org.bson.Document;
import org.bson.json.JsonWriterSettings;
import org.json.JSONObject;
 
public class ChangeStreamExample {
    public static void main(String[] args) {
        MongoClient mongoClient = MongoClients.create("mongodb://<username>:<pwd>@<clustername>.global.mongocluster.cosmos.azure.com/?tls=true");
        MongoDatabase database = mongoClient.getDatabase("test");
        MongoCollection<Document> collection = database.getCollection("test");
 
        // Open a change stream
        MongoCursor<ChangeStreamDocument<Document>> cursor = collection.watch().iterator();
 
        // Listen for changes
        try {
            while (cursor.hasNext()) {
                ChangeStreamDocument<Document> change = cursor.next();
 
                // Create a JSONObject from the fields of the ChangeStreamDocument
                JSONObject json = new JSONObject();
 
                // Add fields to the JSON object
                json.put("operationType", change.getOperationType().getValue());
                json.put("namespace", change.getNamespace().getFullName());
 
                if (change.getDocumentKey() != null) {
                    json.put("documentKey", change.getDocumentKey().toJson().toString());
                }
 
                if (change.getResumeToken() != null) {
                    json.put("resumeToken", change.getResumeToken().toJson());
                }
 
                if (change.getFullDocument() != null) {
                    json.put("fullDocument", change.getFullDocument().toJson());
                }
 
                if (change.getWallTime() != null) {
                    json.put("wallTime", change.getWallTime().toString());
                }
 
                // Pretty-print the JSON object with 4-space indent
                System.out.println(json.toString(4)); 
            }
        } 
        catch (Exception e) {
            System.err.println("Error processing change stream: " + e);
        } 
        finally {
            cursor.close();
            mongoClient.close();
        }
    }
}

Viktigt!

Ändringsströmmar kan återupptas genom att ange en återuppta-token till resumeAfter när markören öppnas. Det förväntas dock att det finns tillräckligt med historik för att hitta åtgärden som är associerad med token. Dokumentet som observerats i ändringsströmmen i _id fältet representerar den återanvändbara token.

cursor = db.exampleCollection.watch(resume_after=resume_token)

Övervaka databasändringar med Change Stream

Nu ska vi förstå utdata från ändringsströmmen via exemplet.

Infoga

I den här ändringsströmshändelsen ser vi att en ny post lades till inserted i samlingen exampleCollection av databasen cs; och händelseinformationen innehåller det fullständiga innehållet i det nyligen tillagda dokumentet.

{
  "_id": { "_data": "AeARBpQ/AAAA" }, // "resume_token"
  "operationType": "insert",
  "fullDocument": {
    "_id": { "$oid": "66e6f63e6f49ecaabf794958" },
    "employee_id": "17986",
    "name": "John",
    "position": "Software Engineer",
    "department": "IT",
    "rating": 4
  },
  "ns": { "db": "cs", "coll": "exampleCollection" },
  "documentKey": { "_id": { "$oid": "66e6f63e6f49ecaabf794958" } }
}

Uppdatering

I den här uppdateringshändelsen ändras position och rating för John. Ändringsströmmen återspeglar en update i exampleCollection samlingen med dokumentets status efter uppdateringen.

{
  "_id": { "_data": "AWACAKM/AAAA" },
  "operationType": "update",
  "fullDocument": {
    "_id": { "$oid": "66e6f63e6f49ecaabf794958" },
    "employee_id": "17986",
    "name": "John",
    "position": "SSE",
    "department": "IT",
    "rating": 5
  },
  "ns": {
    "db": "cs", "coll": "exampleCollection"
  },
  "documentKey": {
    "_id": { "$oid": "66e6f63e6f49ecaabf794958" }
  }
}

Ta bort

Ändringsströmshändelsen anger att ett dokument kom deleted från exampleCollection samlingen i cs databasen. Händelsen fångar den unika identifieraren för det borttagna dokumentet.

{
  "_id": { "_data": "ASgBAJs/AAAA" },
  "operationType": "delete",
  "ns": { "db": "cs", "coll": "exampleCollection" },
  "documentKey": { "_id": { "$oid": "66e6f63e6f49ecaabf794958" } }
}

Anpassa data i Change Stream

Anpassa dina utdata för ändringsströmmen genom att ange en matris med en eller flera pipelinesteg under konfigurationen. De operatörer som stöds är följande.

• $addFields
• $match
• $project
• $set
• $unset

Begränsningar

• Historiska ändringsströmhändelser från den tidigare tidslinjen stöds ännu inte.
• Ändringsströmhändelser i kluster med flera fragment stöds ännu inte.
• Markörer för ändringsström måste återinitieras efter en övergångshändelse i det aktuella tillståndet.
• Update händelsen har ännu inte stöd för uppdateringsbeskrivning.
• pre-image är ett alternativ som inte stöds.
• $changestream som en kapslad pipeline i en annan fas stöds ännu inte.