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

Çekme modelini kullanarak değişiklik akışını işlemek için bir örneği FeedIteratoroluşturun. öğesini ilk 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 bir bölüm anahtarı değerleri aralığıdır ve değişiklik akışından okunabilecek öğeleri belirli FeedIteratorbir kullanılarak 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 belirtecinin değerinden 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. Bu özellik ayarlandığında, 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 öğe sayısı belirtilen değerden yüksek olabilir, böylece aynı işlem tarafından değiştirilen öğeler tek bir atomik toplu işlemin 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);

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, nesneleri döndüren User tüm sürümlerde ve silme modlarında elde etme FeedIterator örneği verilmiştir:

FeedIterator<ChangeFeedItem<User>> InteratorWithPOCOS = container.GetChangeFeedIterator<ChangeFeedItem<User>>(ChangeFeedStartFrom.Now(), ChangeFeedMode.AllVersionsAndDeletes);

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 ilk seri durumdan çıkarmadan okumanıza olanak tanıyarak istemci kaynaklarından tasarruf etmenizi sağlar.

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, çalışma 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 FeedRange tane 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:

1. Makine:

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

2. Makine:

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

Devamlılık 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 . Devamlılık belirteci belirtilirse, 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.

Çekme modelini kullanarak değişiklik akışını işlemek için bir örneği Iterator<FeedResponse<JsonNode>> responseIteratoroluşturun. oluştururken CosmosChangeFeedRequestOptions, değişiklik akışının nereden okunacağını ve kullanmak istediğiniz parametrenin FeedRange geçirileceğini belirtmeniz 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ş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. Bu özellik ayarlandığında, sayfa başına alınan en fazla öğe sayısını ayarlar.

En son sürüm modunda bir responseIterator değer elde 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();

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

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

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

Ardından sonuçları yineleyebiliriz. 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, çalışma 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şiklikleri işlemeyi 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:

1. Makine:

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

2. Makine:

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

Çekme modelini kullanarak değişiklik akışını işlemek için bir örneği ChangeFeedPullModelIteratoroluşturun. öğesini ilk 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.

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. Geçerli zamanda başlayarak tüm değişiklikleri okumaya başlayan bir örnek aşağıda 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ında olduğu 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 FeedRange tane alırsınız.

kullanarak FeedRange, değişiklik akışının birden çok makine veya iş parçacığı arasında işlenmesini paralel hale getirmek 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'i 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:

1. Makine:

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

2. Makine:

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

Devamlılık 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. Devamlılık belirteci belirtilirse, 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);
}