MongoDB için Azure Cosmos DB API'sindeki akışları değiştirme

Makale
06/13/2024

ŞUNLAR IÇIN GEÇERLIDIR: MongoDB

MongoDB için Azure Cosmos DB API'sinde değişiklik akışı desteği, değişiklik akışları API'sini kullanarak kullanılabilir. Değişiklik akışları API'sini kullanarak uygulamalarınız koleksiyonda veya tek parçadaki öğelerde yapılan değişiklikleri alabilir. Daha sonra sonuçlara göre daha fazla işlem yapabilirsiniz. Koleksiyondaki öğelerde yapılan değişiklikler, değişiklik süreleri sırasına göre yakalanır ve parça anahtarı başına sıralama düzeni garanti edilir.

Not

Değişiklik akışlarını kullanmak için Sunucu sürümü 3.6 veya üzeri olan MongoDB hesabı için Azure Cosmos DB API'sini oluşturun. Değişiklik akışı örneklerini önceki bir sürümde çalıştırırsanız Tanınmayan işlem hattı aşama adı: $changeStream hatasını görebilirsiniz.

Örnekler

Aşağıdaki örnekte, koleksiyondaki tüm öğelerde değişiklik akışlarının nasıl alıneceği gösterilmektedir. Bu örnek, eklenen, güncelleştirilen veya değiştirilen öğeleri izlemek için bir imleç oluşturur. Değişiklik $match akışlarını almak için aşama, $project aşama ve fullDocument seçenek gereklidir. Değişiklik akışlarını kullanarak silme işlemlerinin izlenmesi şu anda desteklenmemektedir. Geçici bir çözüm olarak, silinmekte olan öğelere geçici bir işaretçi ekleyebilirsiniz. Örneğin, öğeye "silinmiş" adlı bir öznitelik ekleyebilirsiniz. Öğeyi silmek istediğinizde " silindi" true olarak ayarlayabilir ve öğede bir TTL ayarlayabilirsiniz. "Silindi" olarak güncelleştirilmesi true bir güncelleştirme olduğundan, bu değişiklik değişiklik akışında görünür.

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

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

Aşağıdaki örnekte Java'da değişiklik akışı işlevinin nasıl kullanılacağı gösterilmektedir. Örneğin tamamı için bu GitHub deposuna bakın. Bu örnekte, son okumadaki resumeAfter tüm değişiklikleri aramak için yönteminin nasıl kullanılacağı da gösterilmektedir.

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

Tek bir parça içindeki değişiklikler

Aşağıdaki örnek, tek bir parça içindeki öğelerde nasıl değişiklik yapılacağını gösterir. Bu örnek, parça anahtarı "a" ve parça anahtarı değeri "1" değerine eşit olan öğelerin değişikliklerini alır. Farklı istemcilerin farklı parçalardan gelen değişiklikleri paralel olarak okuması mümkündür.

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

Değişiklik akışlarını ölçeklendirme

Değişiklik akışlarını büyük ölçekte kullanırken yükü eşit bir şekilde yaymak en iyisidir. Yükü fiziksel parçalar/bölümler arasında yaymak için GetChangeStreamTokens özel komutunu kullanın.

Geçerli sınırlamalar

Değişiklik akışları kullanılırken aşağıdaki sınırlamalar geçerlidir:

operationType çıkış belgesinde ve updateDescription özellikleri henüz desteklenmiyor.
insert, updateve replace işlem türleri şu anda desteklenmektedir. Ancak silme işlemi veya diğer olaylar henüz desteklenmiyor.

Bu sınırlamalar nedeniyle, önceki örneklerde gösterildiği gibi $match aşaması, $project aşaması ve fullDocument seçenekleri gereklidir.

NoSQL için Azure Cosmos DB API'sindeki değişiklik akışından farklı olarak, değişiklik akışlarını kullanmak için ayrı bir Değişiklik Akışı İşlemci Kitaplığı veya kira kapsayıcısı gereksinimi yoktur. Şu anda değişiklik akışlarını işlemek için Azure İşlevleri tetikleyicileri desteklenmemektedir.

Hata işleme

Değişiklik akışları kullanılırken aşağıdaki hata kodları ve iletiler desteklenir:

HTTP hata kodu 16500 - Değişiklik akışı kısıtlandığında boş bir sayfa döndürür.
NamespaceNotFound (OperationType Invalidate) - Var olmayan koleksiyonda değişiklik akışı çalıştırırsanız veya koleksiyon bırakılırsa bir NamespaceNotFound hata döndürülür. operationType Özellik çıkış belgesinde döndürülemediğinden hata operationType InvalidateNamespaceNotFound döndürülür.

Aracılığıyla paylaş