Model penarikan umpan perubahan di Azure Cosmos DB

BERLAKU UNTUK: NoSQL

Anda dapat menggunakan model penarikan umpan perubahan untuk menggunakan umpan perubahan Azure Cosmos DB dengan kecepatan Anda sendiri. Mirip dengan prosesor umpan perubahan, Anda dapat menggunakan model penarikan umpan perubahan untuk menyejajarkan pemrosesan perubahan di beberapa konsumen umpan perubahan.

Bandingkan dengan prosesor umpan perubahan

Banyak skenario dapat memproses umpan perubahan dengan menggunakan prosesor umpan perubahan atau model penarikan umpan perubahan. Token kelanjutan model penarikan dan kontainer sewa prosesor umpan perubahan berfungsi sebagai marka buku untuk item terakhir yang diproses atau batch item di umpan perubahan.

Namun, Anda tidak dapat mengonversi token kelanjutan menjadi sewa atau sebaliknya.

Catatan

Dalam kebanyakan kasus, ketika Anda perlu membaca dari umpan perubahan, opsi paling sederhana adalah menggunakan prosesor umpan perubahan.

Anda harus pertimbangkan model penarikan dalam skenario ini:

• Untuk membaca perubahan dari kunci partisi tertentu.
• Untuk mengontrol kecepatan di mana klien Anda menerima perubahan untuk diproses.
• Untuk melakukan pembacaan satu kali data yang ada di umpan perubahan (misalnya, untuk melakukan migrasi data).

Berikut adalah beberapa perbedaan utama antara prosesor umpan perubahan dan model penarikan umpan perubahan:

Fitur	Prosesor umpan perubahan	Model penarikan umpan perubahan
Melacak titik saat ini dalam memproses umpan perubahan	Sewa (disimpan dalam kontainer Azure Cosmos DB)	Token kelanjutan (disimpan dalam memori atau ditahan secara manual)
Kemampuan untuk memutar ulang perubahan sebelumnya	Ya, dengan model pendorongan	Ya, dengan model penarikan
Polling untuk perubahan di masa mendatang	Memeriksa perubahan secara otomatis berdasarkan nilai yang ditentukan WithPollInterval pengguna	Manual
Perilaku tanpa perubahan baru	Tunggu nilai secara otomatis lalu WithPollInterval centang ulang	Harus memeriksa status dan memeriksa ulang secara manual
Memproses perubahan dari seluruh kontainer	Ya, dan secara otomatis diparalelkan di beberapa utas dan mesin yang mengonsumsi dari kontainer yang sama	Ya, dan diparalelkan secara manual dengan menggunakan FeedRange
Memproses perubahan hanya dari satu kunci partisi	Tidak didukung	Ya

Catatan

Saat Anda menggunakan model penarikan, tidak seperti saat membaca dengan menggunakan prosesor umpan perubahan, Anda harus secara eksplisit menangani kasus di mana tidak ada perubahan baru.

Bekerja dengan model penarikan

.NET

Untuk memproses umpan perubahan dengan menggunakan model penarikan, buat instans FeedIterator. Ketika Anda awalnya membuat FeedIterator, Anda harus menentukan nilai yang diperlukan ChangeFeedStartFrom , yang terdiri dari posisi awal untuk membaca perubahan dan nilai yang ingin Anda gunakan untuk FeedRange. FeedRange adalah rentang nilai kunci partisi dan menentukan item yang dapat dibaca dari umpan perubahan dengan menggunakan spesifik FeedIteratortersebut. Anda juga harus menentukan nilai yang diperlukan ChangeFeedMode untuk mode tempat Anda ingin memproses perubahan: versi terbaru atau semua versi dan penghapusan. Gunakan atau ChangeFeedMode.LatestVersionChangeFeedMode.AllVersionsAndDeletes untuk menunjukkan mode mana yang ingin Anda gunakan untuk membaca umpan perubahan. Saat Anda menggunakan semua versi dan mode penghapusan, Anda harus memilih umpan perubahan mulai dari nilai dari Now() atau dari token kelanjutan tertentu.

Anda dapat secara opsional menentukan ChangeFeedRequestOptions untuk mengatur PageSizeHint. Saat diatur, properti ini menetapkan jumlah maksimum item yang diterima per halaman. Jika operasi dalam koleksi yang dipantau dilakukan melalui prosedur tersimpan, cakupan transaksi dipertahankan saat membaca item dari umpan perubahan. Akibatnya, jumlah item yang diterima mungkin lebih tinggi dari nilai yang ditentukan sehingga item yang diubah oleh transaksi yang sama dikembalikan sebagai bagian dari satu batch atomik.

Berikut adalah contoh cara mendapatkan FeedIterator dalam mode versi terbaru yang mengembalikan objek entitas, dalam hal User ini objek:

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

Tip

Sebelum versi 3.34.0, mode versi terbaru dapat digunakan dengan mengatur ChangeFeedMode.Incremental. Baik Incremental dan LatestVersion lihat mode versi terbaru dari umpan perubahan dan aplikasi yang menggunakan salah satu mode akan melihat perilaku yang sama.

Semua versi dan mode penghapusan dalam pratinjau dan dapat digunakan dengan pratinjau versi >.NET SDK = 3.32.0-preview. Berikut adalah contoh untuk mendapatkan FeedIterator di semua versi dan menghapus mode yang mengembalikan objek dinamis:

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

Catatan

Dalam mode versi terbaru, Anda menerima objek yang mewakili item yang berubah, dengan beberapa metadata tambahan. Semua versi dan mode penghapusan mengembalikan model data yang berbeda. Untuk informasi selengkapnya, lihat Mengurai objek respons.

Mengonsumsi umpan perubahan melalui aliran

FeedIterator untuk kedua mode umpan perubahan memiliki dua opsi. Selain contoh yang mengembalikan objek entitas, Anda juga dapat memperoleh respons dengan Stream dukungan. Aliran memungkinkan Anda membaca data tanpa terlebih dahulu deserialisasi, sehingga Anda menghemat sumber daya klien.

Berikut adalah contoh cara mendapatkan FeedIterator dalam mode versi terbaru yang mengembalikan Stream:

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

Mengonsumsi perubahan untuk seluruh kontainer

Jika Anda tidak menyediakan FeedRange parameter ke FeedIterator, Anda dapat memproses seluruh umpan perubahan kontainer dengan kecepatan Anda sendiri. Berikut adalah contoh, yang mulai membaca semua perubahan, dimulai pada waktu saat ini dengan menggunakan mode versi terbaru:

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

Karena umpan perubahan secara efektif merupakan daftar item tak terbatas yang mencakup semua penulisan dan pembaruan di masa mendatang, nilai HasMoreResults selalu true. Ketika Anda mencoba membaca umpan perubahan dan tidak ada perubahan baru yang tersedia, Anda menerima respons dengan NotModified status. Dalam contoh sebelumnya, ini ditangani dengan menunggu lima detik sebelum memeriksa ulang perubahan.

Menggunakan perubahan untuk kunci partisi

Dalam beberapa kasus, Anda mungkin hanya ingin memproses perubahan untuk kunci partisi tertentu. Anda dapat memperoleh FeedIterator untuk kunci partisi tertentu dan memproses perubahan dengan cara yang sama seperti yang Anda dapat untuk seluruh kontainer.

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

Menggunakan FeedRange untuk paralelisasi

Dalam prosesor umpan perubahan, pekerjaan tersebar di beberapa konsumen secara otomatis. Dalam model penarikan umpan perubahan, Anda dapat menggunakan FeedRange untuk paralelisasi pemrosesan umpan perubahan. FeedRange mewakili rentang nilai kunci partisi.

Berikut adalah contoh yang menunjukkan cara mendapatkan daftar rentang untuk kontainer Anda:

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

Saat Anda mendapatkan daftar FeedRange nilai untuk kontainer, Anda mendapatkan satu FeedRange per partisi fisik.

Dengan menggunakan FeedRange, Anda dapat membuat FeedIterator untuk menyejajarkan pemrosesan umpan perubahan di beberapa komputer atau utas. Berbeda dengan contoh sebelumnya yang menunjukkan cara mendapatkan FeedIterator untuk seluruh kontainer atau satu kunci partisi, Anda dapat menggunakan FeedRanges untuk mendapatkan beberapa FeedIterator yang dapat memproses umpan perubahan secara paralel.

Saat ingin menggunakan FeedRanges, Anda harus memiliki proses orkestrator yang mendapatkan FeedRanges dan mendistribusikannya ke mesin tersebut. Distribusi ini mungkin:

• Menggunakan FeedRange.ToJsonString dan mendistribusikan nilai untai (karakter) ini. Konsumen dapat menggunakan nilai ini dengan FeedRange.FromJsonString.
• Jika distribusi sedang berproses, lewati referensi objek FeedRange.

Berikut adalah sampel yang menunjukkan cara membaca dari awal umpan perubahan kontainer dengan menggunakan dua komputer terpisah hipotetis yang dibaca secara paralel:

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

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

Menyimpan token kelanjutan

Anda dapat menyimpan posisi FeedIterator dengan mendapatkan token kelanjutan. Token kelanjutan adalah nilai untai (karakter) yang melacak perubahan terakhir yang diproses FeedIterator Anda dan mengizinkan FeedIterator untuk melanjutkan pada titik ini nantinya. Token kelanjutan, jika ditentukan, lebih diutamakan dari waktu mulai dan mulai dari nilai awal. Kode berikut membaca umpan perubahan sejak pembuatan kontainer. Setelah tidak ada lagi perubahan yang tersedia, kode tersebut akan menahan token kelanjutan sehingga konsumsi umpan perubahan dapat dilanjutkan nanti.

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

Saat Anda menggunakan mode versi terbaru, token kelanjutan FeedIterator tidak pernah kedaluwarsa selama kontainer Azure Cosmos DB masih ada. Saat Anda menggunakan semua versi dan mode penghapusan, FeedIterator token kelanjutan berlaku selama perubahan terjadi dalam jendela retensi untuk pencadangan berkelanjutan.

Java

Untuk memproses umpan perubahan dengan menggunakan model penarikan, buat instans Iterator<FeedResponse<JsonNode>> responseIterator. Saat membuat CosmosChangeFeedRequestOptions, Anda harus menentukan tempat untuk mulai membaca umpan perubahan dari dan meneruskan FeedRange parameter yang ingin Anda gunakan. FeedRange adalah rentang nilai kunci partisi yang menentukan item yang dapat dibaca dari umpan perubahan.

Jika Anda ingin membaca umpan perubahan di semua versi dan mode penghapusan, Anda juga harus menentukan allVersionsAndDeletes() kapan Anda membuat CosmosChangeFeedRequestOptions. Semua versi dan mode penghapusan tidak mendukung pemrosesan umpan perubahan dari awal atau dari titik waktu. Anda harus memproses perubahan dari sekarang atau dari token kelanjutan. Semua versi dan mode penghapusan dalam pratinjau dan tersedia dalam versi >Java SDK = 4.42.0.

Mengonsumsi perubahan untuk seluruh kontainer

Jika Anda menentukan FeedRange.forFullRange(), Anda dapat memproses umpan perubahan untuk seluruh kontainer dengan kecepatan Anda sendiri. Anda dapat secara opsional menentukan nilai dalam byPage(). Saat diatur, properti ini menetapkan jumlah maksimum item yang diterima per halaman.

Catatan

Semua cuplikan kode berikut diambil dari sampel di GitHub. Anda dapat menggunakan sampel mode versi terbaru dan semua versi dan menghapus sampel mode.

Berikut adalah contoh cara mendapatkan responseIterator nilai dalam mode versi terbaru:

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

Berikut adalah contoh cara mendapatkan responseIterator di semua versi dan mode penghapusan:

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

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

Kami kemudian dapat mengulangi hasilnya. Karena umpan perubahan secara efektif merupakan daftar item tak terbatas yang mencakup semua penulisan dan pembaruan di masa mendatang, nilai responseIterator.hasNext() selalu true. Berikut adalah contoh dalam mode versi terbaru, yang membaca semua perubahan, mulai dari awal. Setiap perulangan mempertahankan token kelanjutan setelah memproses semua peristiwa. Ini mengambil dari titik terakhir yang diproses dalam umpan perubahan dan ditangani dengan menggunakan createForProcessingFromContinuation:

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

Mengonsumsi perubahan kunci partisi

Dalam beberapa kasus, Anda mungkin hanya ingin memproses perubahan untuk kunci partisi tertentu. Anda dapat memproses perubahan untuk kunci partisi tertentu dengan cara yang sama seperti yang Anda bisa untuk seluruh kontainer. Berikut adalah contoh yang menggunakan mode versi terbaru:

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

Gunakan FeedRange untuk paralelisasi

Dalam prosesor umpan perubahan, pekerjaan tersebar di beberapa konsumen secara otomatis. Dalam model penarikan umpan perubahan, Anda dapat menggunakan FeedRange untuk paralelisasi pemrosesan umpan perubahan. FeedRange mewakili rentang nilai kunci partisi.

Berikut adalah contoh yang menggunakan mode versi terbaru yang menunjukkan cara mendapatkan daftar rentang untuk kontainer Anda:

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

Ketika Anda mendapatkan daftar FeedRanges untuk kontainer Anda, Anda mendapatkan satu FeedRange per partisi fisik.

Dengan menggunakan FeedRange, Anda dapat menyejajarkan pemrosesan umpan perubahan di beberapa komputer atau utas. Tidak seperti contoh sebelumnya yang menunjukkan cara memproses perubahan untuk seluruh kontainer atau satu kunci partisi, Anda dapat menggunakan FeedRanges untuk memproses feed perubahan secara paralel.

Saat ingin menggunakan FeedRanges, Anda harus memiliki proses orkestrator yang mendapatkan FeedRanges dan mendistribusikannya ke mesin tersebut. Distribusi ini mungkin:

• Menggunakan FeedRange.toString() dan mendistribusikan nilai untai (karakter) ini.
• Jika distribusi sedang berproses, lewati referensi objek FeedRange.

Berikut adalah sampel yang menggunakan mode versi terbaru. Ini menunjukkan cara membaca dari awal umpan perubahan kontainer dengan menggunakan dua komputer terpisah hipotetis yang dibaca secara paralel:

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

Mesin 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

Untuk memproses umpan perubahan dengan menggunakan model penarikan, buat instans ChangeFeedPullModelIterator. Ketika Anda awalnya membuat ChangeFeedPullModelIterator, Anda harus menentukan nilai yang diperlukan changeFeedStartFrom di dalam ChangeFeedIteratorOptions yang terdiri dari posisi awal untuk membaca perubahan dan sumber daya (kunci partisi atau FeedRange) yang perubahannya akan diambil.

Catatan

Jika tidak ada changeFeedStartFrom nilai yang ditentukan, changefeed akan diambil untuk seluruh kontainer dari Now(). Saat ini, hanya versi terbaru yang didukung oleh JS SDK dan dipilih secara default.

Anda dapat secara opsional menggunakan maxItemCount di ChangeFeedIteratorOptions untuk mengatur jumlah maksimum item yang diterima per halaman. Berikut adalah contoh cara mendapatkan iterator dalam mode versi terbaru yang mengembalikan objek entitas:

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

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

Mengonsumsi perubahan untuk seluruh kontainer

Jika Anda tidak menyediakan FeedRange parameter atau PartitionKey di dalam ChangeFeedStartFrom, Anda dapat memproses seluruh umpan perubahan kontainer dengan kecepatan Anda sendiri. Berikut adalah contoh, yang mulai membaca semua perubahan, dimulai pada waktu saat ini:

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

Karena umpan perubahan secara efektif merupakan daftar item tak terbatas yang mencakup semua penulisan dan pembaruan di masa mendatang, nilai hasMoreResults selalu true. Ketika Anda mencoba membaca umpan perubahan dan tidak ada perubahan baru yang tersedia, Anda menerima respons dengan NotModified status. Dalam contoh sebelumnya, ini ditangani dengan menunggu lima detik sebelum memeriksa ulang perubahan.

Menggunakan perubahan untuk kunci partisi

Dalam beberapa kasus, Anda mungkin hanya ingin memproses perubahan untuk kunci partisi tertentu. Anda dapat memperoleh iterator untuk kunci partisi tertentu dan memproses perubahan dengan cara yang sama seperti yang Anda bisa untuk seluruh kontainer.

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

Gunakan FeedRange untuk paralelisasi

Dalam model penarikan umpan perubahan, Anda dapat menggunakan FeedRange untuk paralelisasi pemrosesan umpan perubahan. FeedRange mewakili rentang nilai kunci partisi.

Berikut adalah contoh yang menunjukkan cara mendapatkan daftar rentang untuk kontainer Anda:

const ranges = await container.getFeedRanges();

Saat Anda mendapatkan daftar FeedRange nilai untuk kontainer, Anda mendapatkan satu FeedRange per partisi fisik.

Dengan menggunakan FeedRange, Anda dapat membuat iterator untuk menyejajarkan pemrosesan umpan perubahan di beberapa komputer atau utas. Tidak seperti contoh sebelumnya yang menunjukkan cara mendapatkan iterator changefeed untuk seluruh kontainer atau satu kunci partisi, Anda dapat menggunakan FeedRanges untuk mendapatkan beberapa iterator, yang dapat memproses umpan perubahan secara paralel.

Berikut adalah sampel yang menunjukkan cara membaca dari awal umpan perubahan kontainer dengan menggunakan dua komputer terpisah hipotetis yang dibaca secara paralel:

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

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

Menyimpan token kelanjutan

Anda dapat menyimpan posisi iterator Anda dengan mendapatkan token kelanjutan. Token kelanjutan adalah nilai string yang melacak perubahan iterator changefeed Terakhir yang diproses dan memungkinkan iterator dilanjutkan pada titik ini nanti. Token kelanjutan, jika ditentukan, lebih diutamakan daripada waktu mulai dan dimulai dari nilai awal. Kode berikut membaca umpan perubahan sejak pembuatan kontainer. Setelah tidak ada lagi perubahan yang tersedia, kode tersebut akan menahan token kelanjutan sehingga konsumsi umpan perubahan dapat dilanjutkan nanti.

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

Token kelanjutan tidak pernah kedaluwarsa selama kontainer Azure Cosmos DB masih ada.

Gunakan AsinkronIterator

Anda dapat menggunakan JavaScript Async Iterator untuk mengambil umpan perubahan. Berikut adalah contoh untuk menggunakan Iterator Asinkron.

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

Langkah berikutnya

• Ikhtisar umpan perubahan
• Menggunakan prosesor umpan perubahan
• Pemicu Azure Functions