Azure Cosmos DB'de akış çekme modelini değiştirme

UYGULANANLAR: NOSQL

Azure Cosmos DB değişiklik akışını kendi hızınızda kullanmak için değişiklik akışı çekme modelini kullanabilirsiniz. Değişiklik akışı işlemcisine benzer şekilde, değişikliklerin birden çok değişiklik akışı tüketicisinde işlenmesini paralelleştirmek için değişiklik akışı çekme modelini kullanabilirsiniz.

Değişiklik akışı işlemcisi ile karşılaştırma

Birçok senaryo, değişiklik akışı işlemcisini veya değişiklik akışı çekme modelini kullanarak değişiklik akışını işleyebilir. Çekme modelinin devamlılık belirteçleri ve değişiklik akışı işlemcisinin kira kapsayıcısı, hem son işlenen öğe için yer işaretleri hem de değişiklik akışındaki öğeler grubu olarak çalışır.

Ancak, devamlılık belirteçlerini kiraya dönüştüremezsiniz veya bunun tersi de geçerlidir.

Not

Çoğu durumda, değişiklik akışından okumanız gerektiğinde, en basit seçenek değişiklik akışı işlemcisini kullanmaktır.

Çekme modelini şu senaryolarda kullanmayı düşünmelisiniz:

• Belirli bir bölüm anahtarındaki değişiklikleri okumak için.
• İstemcinizin işlenmek üzere değişiklik alma hızını denetlemek için.
• Değişiklik akışındaki mevcut verilerin tek seferlik okunmasını gerçekleştirmek için (örneğin, veri geçişi yapmak için).

Değişiklik akışı işlemcisi ile değişiklik akışı çekme modeli arasındaki bazı önemli farklar şunlardır:

Özellik	Değişiklik akışı işlemcisi	Değişiklik akışı çekme modeli
Değişiklik akışını işlemedeki geçerli noktayı izleme	Kira (Azure Cosmos DB kapsayıcısında depolanır)	Devamlılık belirteci (bellekte depolanır veya el ile kalıcı hale gelir)
Geçmiş değişiklikleri yeniden yürütme olanağı	Evet, gönderme modeliyle	Evet, çekme modeliyle
Gelecekteki değişiklikler için yoklama	Kullanıcı tarafından belirtilen WithPollInterval değere göre değişiklikleri otomatik olarak denetler	El ile
Yeni değişiklik yapılmadığı durumlarda davranış	değerini WithPollInterval otomatik olarak bekleyin ve ardından yeniden denetleyin	Durumu denetlemeli ve el ile yeniden denetlemelidir
Kapsayıcının tamamından değişiklikleri işleme	Evet ve aynı kapsayıcıdan kullanan birden çok iş parçacığı ve makine arasında otomatik olarak paralelleştirilmiştir	Evet ve kullanarak el ile paralelleştirilmiş FeedRange
Değişiklikleri yalnızca tek bir bölüm anahtarından işleme	Desteklenmez	Yes

Not

Çekme modelini kullanırken, değişiklik akışı işlemcisini kullanarak okumanın aksine, yeni değişiklik olmayan durumları açıkça işlemeniz gerekir.

Çekme modeliyle çalışma

.NET

Çekme modelini kullanarak değişiklik akışını işlemek için bir örneği FeedIteratoroluşturun. İlk oluşturduğunuzdaFeedIterator, hem değişiklikleri okumak için başlangıç konumundan hem de için FeedRangekullanmak istediğiniz değerden oluşan gerekli ChangeFeedStartFrom bir değer belirtmeniz gerekir. FeedRange bölüm anahtarı değerleri aralığıdır ve değişiklik akışından okunabilecek öğeleri belirli FeedIteratorbir kullanarak belirtir. Değişiklikleri işlemek istediğiniz mod için de gerekli ChangeFeedMode bir değer belirtmeniz gerekir: en son sürüm veya tüm sürümler ve silmeler. ChangeFeedMode.LatestVersion Değişiklik akışını okumak için hangi modu kullanmak istediğinizi belirtmek için veya ChangeFeedMode.AllVersionsAndDeletes kullanın. Tüm sürümleri ve silme modunu kullandığınızda, belirli bir devamlılık belirtecinden veya değerinden Now() bir değişiklik akışı başlangıcı seçmeniz gerekir.

İsteğe bağlı olarak bir PageSizeHintayarlamak için belirtebilirsinizChangeFeedRequestOptions. Ayarlandığında, bu özellik sayfa başına alınan en fazla öğe sayısını ayarlar. İzlenen koleksiyondaki işlemler saklı yordamlar aracılığıyla gerçekleştiriliyorsa, değişiklik akışındaki öğeler okunurken işlem kapsamı korunur. Sonuç olarak, alınan öğelerin sayısı belirtilen değerden yüksek olabilir, böylece aynı işlem tarafından değiştirilen öğeler tek bir atomik toplu işin parçası olarak döndürülür.

Burada, varlık nesnelerini döndüren en son sürüm modunda nasıl edinildiğini FeedIterator gösteren bir örnek verilmiştir. Bu örnekte bir User nesne verilmiştir:

FeedIterator<User> InteratorWithPOCOS = container.GetChangeFeedIterator<User>(ChangeFeedStartFrom.Beginning(), ChangeFeedMode.LatestVersion);

İpucu

sürümünden 3.34.0önce, ayarıyla ChangeFeedMode.Incrementalen son sürüm modu kullanılabilir. LatestVersion Hem hem de Incremental değişiklik akışının en son sürüm moduna bakın ve her iki modu kullanan uygulamalar da aynı davranışı görür.

Tüm sürümler ve silmeler modu önizleme aşamasındadır ve önizleme .NET SDK sürümleri >= 3.32.0-previewile kullanılabilir. Aşağıda, dinamik nesneler döndüren tüm sürümler ve silmeler modunda elde etme FeedIterator örneği verilmiştir:

FeedIterator<dynamic> InteratorWithDynamic = container.GetChangeFeedIterator<dynamic>(ChangeFeedStartFrom.Now(), ChangeFeedMode.AllVersionsAndDeletes);

Not

En son sürüm modunda, değiştirilen öğeyi temsil eden nesneleri ve bazı ek meta verileri alırsınız. Tüm sürümler ve silmeler modu farklı bir veri modeli döndürür. Daha fazla bilgi için bkz. Yanıt nesnesini ayrıştırma.

Akışlar aracılığıyla değişiklik akışını kullanma

FeedIterator her iki değişiklik akışı modu için iki seçenek vardır. Varlık nesnelerini döndüren örneklere ek olarak, yanıtı destekle Stream de alabilirsiniz. Akışlar, verileri önce seri durumdan çıkarmadan okumanıza olanak sağlar, bu nedenle istemci kaynaklarına kaydedersiniz.

Aşağıda, döndüren en son sürüm modunda nasıl edinildiğini FeedIterator gösteren bir örnek verilmişti Stream:

FeedIterator iteratorWithStreams = container.GetChangeFeedStreamIterator(ChangeFeedStartFrom.Beginning(), ChangeFeedMode.LatestVersion);

Kapsayıcının tamamı için değişiklikleri kullanma

parametresini FeedIteratorsağlamazsanızFeedRange, kapsayıcının değişiklik akışının tamamını kendi hızınızda işleyebilirsiniz. Aşağıda, en son sürüm modunu kullanarak geçerli zamanda başlayarak tüm değişiklikleri okumaya başlayan bir örnek verilmişti:

FeedIterator<User> iteratorForTheEntireContainer = container.GetChangeFeedIterator<User>(ChangeFeedStartFrom.Now(), ChangeFeedMode.LatestVersion);

while (iteratorForTheEntireContainer.HasMoreResults)
{
    FeedResponse<User> response = await iteratorForTheEntireContainer.ReadNextAsync();

    if (response.StatusCode == HttpStatusCode.NotModified)
    {
        Console.WriteLine($"No new changes");
        await Task.Delay(TimeSpan.FromSeconds(5));
    }
    else 
    {
        foreach (User user in response)
        {
            Console.WriteLine($"Detected change for user with id {user.id}");
        }
    }
}

Değişiklik akışı, gelecekteki tüm yazma ve güncelleştirmeleri kapsayan sonsuz bir öğe listesi olduğundan, değeri HasMoreResults her zaman trueolur. Değişiklik akışını okumaya çalıştığınızda ve kullanılabilir yeni bir değişiklik olmadığında, durum bilgisi olan NotModified bir yanıt alırsınız. Yukarıdaki örnekte, değişiklikler yeniden denetlenmeden önce beş saniye beklenerek işlenir.

Bölüm anahtarı için değişiklikleri kullanma

Bazı durumlarda, yalnızca belirli bir bölüm anahtarına yönelik değişiklikleri işlemek isteyebilirsiniz. Belirli bir bölüm anahtarı için alabilir FeedIterator ve değişiklikleri kapsayıcının tamamında yaptığınız gibi işleyebilirsiniz.

FeedIterator<User> iteratorForPartitionKey = container.GetChangeFeedIterator<User>(
    ChangeFeedStartFrom.Beginning(FeedRange.FromPartitionKey(new PartitionKey("PartitionKeyValue")), ChangeFeedMode.LatestVersion));

while (iteratorForThePartitionKey.HasMoreResults)
{
    FeedResponse<User> response = await iteratorForThePartitionKey.ReadNextAsync();

    if (response.StatusCode == HttpStatusCode.NotModified)
    {
        Console.WriteLine($"No new changes");
        await Task.Delay(TimeSpan.FromSeconds(5));
    }
    else
    {
        foreach (User user in response)
        {
            Console.WriteLine($"Detected change for user with id {user.id}");
        }
    }
}

Paralelleştirme için FeedRange kullanma

Değişiklik akışı işlemcisinde iş otomatik olarak birden çok tüketiciye yayılır. Değişiklik akışı çekme modelinde, değişiklik akışının işlenmesini paralelleştirmek için öğesini kullanabilirsiniz FeedRange . A FeedRange , bölüm anahtarı değerleri aralığını temsil eder.

Kapsayıcınız için aralıkların listesinin nasıl alındığını gösteren bir örnek aşağıda verilmiştir:

IReadOnlyList<FeedRange> ranges = await container.GetFeedRangesAsync();

Kapsayıcınız için değerlerin FeedRange listesini aldığınızda, fiziksel bölüm başına bir tane FeedRange alırsınız.

kullanarak FeedRange, değişiklik akışının birden çok makine veya iş parçacığı arasında işlenmesini paralelleştirmek için bir FeedIterator oluşturabilirsiniz. Kapsayıcının tamamı için veya tek bir bölüm anahtarının nasıl alındığını FeedIterator gösteren önceki örnekten farklı olarak, değişiklik akışını paralel olarak işleyebilen birden çok FeedIterator elde etmek için FeedRanges kullanabilirsiniz.

FeedRanges kullanmak istediğiniz durumlarda, FeedRanges'i alan ve bunları bu makinelere dağıtan bir düzenleyici işlemine sahip olmanız gerekir. Bu dağıtım şöyle olabilir:

• Bu dize değerini kullanma FeedRange.ToJsonString ve dağıtma. Tüketiciler bu değeri ile FeedRange.FromJsonStringkullanabilir.
• Dağıtım devam ediyorsa, nesne başvuruyu FeedRange geçirin.

Aşağıda, paralel olarak okunan iki varsayımsal ayrı makine kullanılarak kapsayıcının değişiklik akışının başından itibaren nasıl okunduğunu gösteren bir örnek verilmiştir:

Makine 1:

FeedIterator<User> iteratorA = container.GetChangeFeedIterator<User>(ChangeFeedStartFrom.Beginning(ranges[0]), ChangeFeedMode.LatestVersion);
while (iteratorA.HasMoreResults)
{
    FeedResponse<User> response = await iteratorA.ReadNextAsync();

    if (response.StatusCode == HttpStatusCode.NotModified)
    {
        Console.WriteLine($"No new changes");
        await Task.Delay(TimeSpan.FromSeconds(5));
    }
    else
    {
        foreach (User user in response)
        {
            Console.WriteLine($"Detected change for user with id {user.id}");
        }
    }
}

Makine 2:

FeedIterator<User> iteratorB = container.GetChangeFeedIterator<User>(ChangeFeedStartFrom.Beginning(ranges[1]), ChangeFeedMode.LatestVersion);
while (iteratorB.HasMoreResults)
{
    FeedResponse<User> response = await iteratorA.ReadNextAsync();

    if (response.StatusCode == HttpStatusCode.NotModified)
    {
        Console.WriteLine($"No new changes");
        await Task.Delay(TimeSpan.FromSeconds(5));
    }
    else
    {
        foreach (User user in response)
        {
            Console.WriteLine($"Detected change for user with id {user.id}");
        }
    }
}

Devam belirteçlerini kaydetme

Devamlılık belirtecini alarak konumunuzu FeedIterator kaydedebilirsiniz. Devamlılık belirteci, FeedIterator'ınızın son işlenen değişikliklerini izleyen ve daha sonra bu noktada sürdürülmesini sağlayan bir dize değeridir FeedIterator . Belirtilirse devamlılık belirteci başlangıç zamanından önceliklidir ve başlangıç değerlerinden başlar. Aşağıdaki kod, kapsayıcı oluşturma işleminden bu yana değişiklik akışını okur. Daha fazla değişiklik sağlandıktan sonra, değişiklik akışı tüketiminin daha sonra sürdürülebilmesi için bir devamlılık belirteci kalıcı hale gelir.

FeedIterator<User> iterator = container.GetChangeFeedIterator<User>(ChangeFeedStartFrom.Beginning(), ChangeFeedMode.LatestVersion);

string continuation = null;

while (iterator.HasMoreResults)
{
    FeedResponse<User> response = await iterator.ReadNextAsync();

    if (response.StatusCode == HttpStatusCode.NotModified)
    {
        Console.WriteLine($"No new changes");
        continuation = response.ContinuationToken;
        // Stop the consumption since there are no new changes
        break;
    }
    else
    {
        foreach (User user in response)
        {
            Console.WriteLine($"Detected change for user with id {user.id}");
        }
    }
}

// Some time later when I want to check changes again
FeedIterator<User> iteratorThatResumesFromLastPoint = container.GetChangeFeedIterator<User>(ChangeFeedStartFrom.ContinuationToken(continuation), ChangeFeedMode.LatestVersion);

En son sürüm modunu kullanırken, Azure Cosmos DB kapsayıcısı FeedIterator hala mevcut olduğu sürece devamlılık belirtecinin süresi hiçbir zaman dolmaz. Tüm sürümleri ve silme modunu kullanırken, FeedIterator sürekli yedeklemeler için bekletme penceresinde değişiklikler olduğu sürece devamlılık belirteci geçerli olur.

Java

Çekme modelini kullanarak değişiklik akışını işlemek için bir örneği Iterator<FeedResponse<JsonNode>> responseIteratoroluşturun. Oluşturduğunuzda CosmosChangeFeedRequestOptions, değişiklik akışını okumaya nereden başlayacağınızı belirtmeniz ve kullanmak istediğiniz parametreyi FeedRange geçirmeniz gerekir. FeedRange, değişiklik akışından okunabilecek öğeleri belirten bir bölüm anahtarı değerleri aralığıdır.

Değişiklik akışını tüm sürümlerde ve silme modunda okumak istiyorsanız, öğesini ne zaman oluşturduğunuzu CosmosChangeFeedRequestOptionsda belirtmeniz allVersionsAndDeletes() gerekir. Tüm sürümler ve silmeler modu, değişiklik akışının başlangıçtan veya belirli bir noktadan işlenmesini desteklemez. Değişiklikleri şu andan itibaren veya bir devamlılık belirtecinden işlemeniz gerekir. Tüm sürümler ve silmeler modu önizleme aşamasındadır ve Java SDK sürümü >= 4.42.0ile kullanılabilir.

Kapsayıcının tamamı için değişiklikleri kullanma

belirtirseniz FeedRange.forFullRange(), kapsayıcının tamamı için değişiklik akışını kendi hızınızda işleyebilirsiniz. İsteğe bağlı olarak içinde byPage()bir değer belirtebilirsiniz. Ayarlandığında, bu özellik sayfa başına alınan en fazla öğe sayısını ayarlar.

Not

Aşağıdaki kod parçacıklarının tümü GitHub'daki örneklerden alınmıştır. En son sürüm modu örneğini ve tüm sürümler ve silme modu örneğini kullanabilirsiniz.

En son sürüm modunda bir değer elde responseIterator etme örneği aşağıda verilmişti:

CosmosChangeFeedRequestOptions options = CosmosChangeFeedRequestOptions
        .createForProcessingFromBeginning(FeedRange.forFullRange());
Iterator<FeedResponse<JsonNode>> responseIterator = container
    .queryChangeFeed(options, JsonNode.class)
    .byPage()
    .toIterable()
    .iterator();

Aşağıda, tüm sürümlerde ve silme modunda nasıl responseIterator edineceğinizi gösteren bir örnek verilmiştir:

CosmosChangeFeedRequestOptions options = CosmosChangeFeedRequestOptions
    .createForProcessingFromNow(FeedRange.forFullRange())
    .allVersionsAndDeletes();

Iterator<FeedResponse<JsonNode>> responseIterator = container
    .queryChangeFeed(options, JsonNode.class)
    .byPage()
    .toIterable()
    .iterator();

Daha sonra sonuçlar üzerinde yineleme yapabiliriz. Değişiklik akışı, gelecekteki tüm yazma ve güncelleştirmeleri kapsayan sonsuz bir öğe listesi olduğundan, değeri responseIterator.hasNext() her zaman trueolur. En son sürüm modunda, baştan başlayarak tüm değişiklikleri okuyan bir örnek aşağıda verilmiştir. Her yineleme, tüm olayları işledikten sonra bir devamlılık belirteci kalıcı hale gelir. Değişiklik akışındaki son işlenen noktadan alır ve kullanılarak createForProcessingFromContinuationişlenir:

int i = 0;
List<JsonNode> results;
while (responseIterator.hasNext()) {
    FeedResponse<JsonNode> response = responseIterator.next();
    results = response.getResults();
    logger.info("Got " + results.size() + " items(s)");

    // applying the continuation token
    // only after processing all events
    options = CosmosChangeFeedRequestOptions
            .createForProcessingFromContinuation(response.getContinuationToken());
    i++;
    if (i >= 5) {
        // artificially breaking out of loop - not required in a real app
        System.out.println("breaking....");
        break;
    }
}

Bölüm anahtarının değişikliklerini kullanma

Bazı durumlarda, yalnızca belirli bir bölüm anahtarına yönelik değişiklikleri işlemek isteyebilirsiniz. Belirli bir bölüm anahtarındaki değişiklikleri kapsayıcının tamamında yaptığınız gibi işleyebilirsiniz. İşte en son sürüm modunu kullanan bir örnek:

options = CosmosChangeFeedRequestOptions
        .createForProcessingFromBeginning(FeedRange.forLogicalPartition(new PartitionKey(partitionKey)));

responseIterator = container
    .queryChangeFeed(options, JsonNode.class)
    .byPage()
    .toIterable()
    .iterator();

int pkIndex = 0;

while (responseIterator.hasNext()) {
    FeedResponse<JsonNode> response = responseIterator.next();
    results = response.getResults();
    logger.info("Got " + results.size() + " items(s) retrieved");

    // applying the continuation token
    // only after processing all events
    options = CosmosChangeFeedRequestOptions
        .createForProcessingFromContinuation(response.getContinuationToken());
    pkIndex++;
    if (pkIndex >= 5) {
        // artificially breaking out of loop
        System.out.println("breaking....");
        break;
    }
}

Paralelleştirme için FeedRange kullanma

Değişiklik akışı işlemcisinde iş otomatik olarak birden çok tüketiciye yayılır. Değişiklik akışı çekme modelinde, değişiklik akışının işlenmesini paralelleştirmek için öğesini kullanabilirsiniz FeedRange . A FeedRange , bölüm anahtarı değerleri aralığını temsil eder.

Kapsayıcınız için aralık listesinin nasıl alındığını gösteren en son sürüm modunu kullanan bir örnek aşağıda verilmiştir:

Mono<List<FeedRange>> feedranges = resources.container.getFeedRanges();
List<FeedRange> feedRangeList = feedranges.block();

Kapsayıcınız için FeedRanges listesini aldığınızda, fiziksel bölüm başına bir tane FeedRange alırsınız.

kullanarak FeedRange, değişiklik akışını birden çok makine veya iş parçacığı arasında işlemeyi paralel hale getirebilirsiniz. Kapsayıcının tamamı veya tek bir bölüm anahtarı için değişikliklerin nasıl işlendiğini gösteren önceki örnekten farklı olarak, değişiklik akışını paralel olarak işlemek için FeedRanges'i kullanabilirsiniz.

FeedRanges kullanmak istediğiniz durumlarda, FeedRanges'i alan ve bunları bu makinelere dağıtan bir düzenleyici işlemine sahip olmanız gerekir. Bu dağıtım şöyle olabilir:

• Bu dize değerini kullanma FeedRange.toString() ve dağıtma.
• Dağıtım devam ediyorsa, nesne başvuruyu FeedRange geçirin.

İşte en son sürüm modunu kullanan bir örnek. Paralel olarak okunan iki varsayımsal ayrı makine kullanarak kapsayıcının değişiklik akışının başından itibaren nasıl okunduğunu gösterir:

Makine 1:

FeedRange range1 = feedRangeList.get(0);
options = CosmosChangeFeedRequestOptions
        .createForProcessingFromBeginning(range1);

int machine1index = 0;
responseIterator = container
    .queryChangeFeed(options, JsonNode.class)
    .byPage()
    .toIterable()
    .iterator();

while (responseIterator.hasNext()) {
    FeedResponse<JsonNode> response = responseIterator.next();
    results = response.getResults();
    logger.info("Got " + results.size() + " items(s) retrieved");

    // applying the continuation token
    // only after processing all events
    options = CosmosChangeFeedRequestOptions
        .createForProcessingFromContinuation(response.getContinuationToken());

    machine1index++;

    if (machine1index >= 5) {
        // artificially breaking out of loop - not required in a real app
        System.out.println("breaking....");
        break;
    }
}

Makine 2:

FeedRange range2 = feedRangeList.get(1);
options = CosmosChangeFeedRequestOptions
        .createForProcessingFromBeginning(range2);

responseIterator = container
    .queryChangeFeed(options, JsonNode.class)
    .byPage()
    .toIterable()
    .iterator();

int machine2index = 0;

while (responseIterator.hasNext()) {
    FeedResponse<JsonNode> response = responseIterator.next();
    results = response.getResults();
    logger.info("Got " + results.size() + " items(s) retrieved");

    // applying the continuation token
    // only after processing all events
    options = CosmosChangeFeedRequestOptions
        .createForProcessingFromContinuation(response.getContinuationToken());

    machine2index++;
    if (machine2index >= 5) {
        // artificially breaking out of loop - not required in a real app
        System.out.println("breaking....");
        break;
    }
}

JavaScript

Çekme modelini kullanarak değişiklik akışını işlemek için bir örneği ChangeFeedPullModelIteratoroluşturun. başlangıçta oluşturduğunuzdaChangeFeedPullModelIterator, içinde ChangeFeedIteratorOptions hem değişiklikleri okumak için başlangıç konumunu hem de değişikliklerin getirileceği kaynağı (bölüm anahtarı veya FeedRange) içeren gerekli changeFeedStartFrom bir değer belirtmeniz gerekir.

Not

Hiçbir changeFeedStartFrom değer belirtilmezse, Now() içinden kapsayıcının tamamı için değişiklik akışı getirilir. Şu anda yalnızca en son sürüm JS SDK tarafından desteklenir ve varsayılan olarak seçilidir.

sayfa başına alınan en fazla öğe sayısını ayarlamak için isteğe bağlı olarak içinde ChangeFeedIteratorOptions kullanabilirsinizmaxItemCount. Burada, varlık nesnelerini döndüren en son sürüm modunda yineleyicinin nasıl alındığını gösteren bir örnek verilmiştir:

const options = {
    changeFeedStartFrom: ChangeFeedStartFrom.Now()
};

const iterator = container.items.getChangeFeedIterator(options);

Kapsayıcının tamamı için değişiklikleri kullanma

içinde ChangeFeedStartFrombir FeedRange veya PartitionKey parametresi sağlamazsanız, kapsayıcının değişiklik akışının tamamını kendi hızınızda işleyebilirsiniz. Aşağıda, geçerli zamanda başlayarak tüm değişiklikleri okumaya başlayan bir örnek verilmişti:

async function waitFor(milliseconds: number): Promise<void> {
  return new Promise((resolve) => setTimeout(resolve, milliseconds));
}

const options = {
      changeFeedStartFrom: ChangeFeedStartFrom.Beginning()
};

const iterator = container.items.getChangeFeedIterator(options);

let timeout = 0;

while(iterator.hasMoreResults) {
    const response = await iterator.readNext();
    if (response.statusCode === StatusCodes.NotModified) {
        timeout = 5000;
    } 
    else {
        console.log("Result found", response.result);
        timeout = 0;
    }
    await waitFor(timeout);
}

Değişiklik akışı, gelecekteki tüm yazma ve güncelleştirmeleri kapsayan sonsuz bir öğe listesi olduğundan, değeri hasMoreResults her zaman trueolur. Değişiklik akışını okumaya çalıştığınızda ve kullanılabilir yeni bir değişiklik olmadığında, durum bilgisi olan NotModified bir yanıt alırsınız. Yukarıdaki örnekte, değişiklikler yeniden denetlenmeden önce beş saniye beklenerek işlenir.

Bölüm anahtarı için değişiklikleri kullanma

Bazı durumlarda, yalnızca belirli bir bölüm anahtarına yönelik değişiklikleri işlemek isteyebilirsiniz. Belirli bir bölüm anahtarı için yineleyici alabilir ve değişiklikleri kapsayıcının tamamı için yaptığınız gibi işleyebilirsiniz.

async function waitFor(milliseconds: number): Promise<void> {
  return new Promise((resolve) => setTimeout(resolve, milliseconds));
}

const options = {
      changeFeedStartFrom: ChangeFeedStartFrom.Beginning("partitionKeyValue")
};

const iterator = container.items.getChangeFeedIterator(options);

let timeout = 0;

while(iterator.hasMoreResults) {
    const response = await iterator.readNext();
    if (response.statusCode === StatusCodes.NotModified) {
        timeout = 5000;
    } 
    else {
        console.log("Result found", response.result);
        timeout = 0;
    }
    await waitFor(timeout);
}

Paralelleştirme için FeedRange kullanma

Değişiklik akışı çekme modelinde, değişiklik akışının işlenmesini paralelleştirmek için öğesini kullanabilirsiniz FeedRange . A FeedRange , bölüm anahtarı değerleri aralığını temsil eder.

Kapsayıcınız için aralıkların listesinin nasıl alındığını gösteren bir örnek aşağıda verilmiştir:

const ranges = await container.getFeedRanges();

Kapsayıcınız için değerlerin FeedRange listesini aldığınızda, fiziksel bölüm başına bir tane FeedRange alırsınız.

kullanarak FeedRange, değişiklik akışının birden çok makine veya iş parçacığı arasında işlenmesini paralelleştirmek için yineleyici oluşturabilirsiniz. Kapsayıcının tamamı veya tek bir bölüm anahtarı için değişiklik akışı yineleyicisinin nasıl alındığını gösteren önceki örnekten farklı olarak, değişiklik akışını paralel olarak işleyebilen birden çok yineleyici elde etmek için FeedRanges kullanabilirsiniz.

Aşağıda, paralel olarak okunan iki varsayımsal ayrı makine kullanılarak kapsayıcının değişiklik akışının başından itibaren nasıl okunduğunu gösteren bir örnek verilmiştir:

Makine 1:

async function waitFor(milliseconds: number): Promise<void> {
  return new Promise((resolve) => setTimeout(resolve, milliseconds));
}

const options = {
      changeFeedStartFrom: ChangeFeedStartFrom.Beginning(ranges[0])
};

const iterator = container.items.getChangeFeedIterator(options);

let timeout = 0;

while(iterator.hasMoreResults) {
    const response = await iterator.readNext();
    if (response.statusCode === StatusCodes.NotModified) {
        timeout = 5000;
    } 
    else {
        console.log("Result found", response.result);
        timeout = 0;
    }
    await waitFor(timeout);
}

Makine 2:

async function waitFor(milliseconds: number): Promise<void> {
  return new Promise((resolve) => setTimeout(resolve, milliseconds));
}

const options = {
      changeFeedStartFrom: ChangeFeedStartFrom.Beginning(ranges[1])
};

const iterator = container.items.getChangeFeedIterator(options);

let timeout = 0;

while(iterator.hasMoreResults) {
    const response = await iterator.readNext();
    if (response.statusCode === StatusCodes.NotModified) {
        timeout = 5000;
    } 
    else {
        console.log("Result found", response.result);
        timeout = 0;
    }
    await waitFor(timeout);
}

Devam belirteçlerini kaydetme

Devam belirtecini alarak yineleyicinizin konumunu kaydedebilirsiniz. Devamlılık belirteci, değişiklik akışı yineleyicinizin son işlenen değişikliklerini izleyen ve yineleyicinin daha sonra bu noktada sürdürülmesini sağlayan bir dize değeridir. Belirtilirse devamlılık belirteci başlangıç zamanından önceliklidir ve başlangıç değerlerinden başlar. Aşağıdaki kod, kapsayıcı oluşturma işleminden bu yana değişiklik akışını okur. Daha fazla değişiklik sağlandıktan sonra, değişiklik akışı tüketiminin daha sonra sürdürülebilmesi için bir devamlılık belirteci kalıcı hale gelir.

const options = {
      changeFeedStartFrom: ChangeFeedStartFrom.Beginning()
};

const iterator = container.items.getChangeFeedIterator(options);

let timeout = 0;
let continuation = "";
while(iterator.hasMoreResults) {
    const response = await iterator.readNext();
    if (response.statusCode === StatusCodes.NotModified) {
        continuation = response.continuationToken;
        break;
    } 
    else {
        console.log("Result found", response.result);
    }
}

// For checking any new changes using the continuation token
const continuationOptions = {
    changeFeedStartFrom: ChangeFeedStartFrom(continuation)
}
const newIterator = container.items.getChangeFeedIterator(continuationOptions);

Azure Cosmos DB kapsayıcısı hala mevcut olduğu sürece devamlılık belirtecinin süresi hiçbir zaman dolmaz.

AsyncIterator kullanma

Değişiklik akışını getirmek için JavaScript Zaman Uyumsuz Yineleyicisi'ni kullanabilirsiniz. Zaman Uyumsuz Yineleyici'yi kullanmaya yönelik bir örnek aşağıda verilmiştir.

async function waitFor(milliseconds: number): Promise<void> {
  return new Promise((resolve) => setTimeout(resolve, milliseconds));
}
const options = {
      changeFeedStartFrom: ChangeFeedStartFrom.Beginning()
};
let timeout = 0;

for await(const result of container.items.getChangeFeedIterator(options).getAsyncIterator()) {
    if (result.statusCode === StatusCodes.NotModified) {
      timeout = 5000;
    }
    else {
      console.log("Result found", result.result);
      timeout = 0;
    }
    await waitFor(timeout);
}

Sonraki adımlar

• Değişiklik akışına genel bakış
• Değişiklik akışı işlemcisini kullanma
• Azure İşlevleri'ni tetikleme