Változáscsatorna-lekérési modell az Azure Cosmos DB-ben

A KÖVETKEZŐKRE VONATKOZIK: NoSQL

A változáscsatorna lekérési modelljét használva saját tempójában használhatja fel az Azure Cosmos DB változáscsatornáját. A változáscsatorna-feldolgozóhoz hasonlóan a változáscsatorna lekérési modelljével párhuzamosíthatja a módosítások feldolgozását több változáscsatorna-fogyasztó között.

Összehasonlítás a változáscsatorna-feldolgozóval

Számos forgatókönyv képes feldolgozni a változáscsatornát a változáscsatorna feldolgozójának vagy a változáscsatorna lekérési modelljének használatával. A lekérési modell folytatási jogkivonatai és a változáscsatorna feldolgozójának bérlettárolója egyaránt könyvjelzőként működik a változáscsatorna utolsó feldolgozott eleméhez vagy tételkötegéhez.

A folytatási jogkivonatokat azonban nem konvertálhatja bérletté, vagy fordítva.

Megjegyzés

A legtöbb esetben, amikor a változáscsatornából kell olvasnia, a legegyszerűbb megoldás a változáscsatorna feldolgozójának használata.

Az alábbi esetekben érdemes megfontolnia a lekéréses modell használatát:

• Egy adott partíciókulcs módosításainak beolvasása.
• Annak szabályozása, hogy az ügyfél milyen ütemben fogadja a módosításokat a feldolgozáshoz.
• A változáscsatorna meglévő adatainak egyszeri olvasásához (például adatmigráláshoz).

Íme néhány fő különbség a változáscsatorna feldolgozója és a változáscsatorna lekérési modellje között:

Szolgáltatás	Változáscsatorna feldolgozója	Változáscsatorna lekérési modellje
A változáscsatorna feldolgozásának aktuális pontjának nyomon követése	Bérlet (egy Azure Cosmos DB-tárolóban tárolva)	Folytatási jogkivonat (a memóriában tárolva vagy manuálisan megőrzve)
A korábbi módosítások visszajátszásának lehetősége	Igen, leküldéses modellel	Igen, lekéréses modellel
Lekérdezés a jövőbeli változásokról	Automatikusan ellenőrzi a felhasználó által megadott WithPollInterval értéken alapuló módosításokat	Kézi
Viselkedés, ahol nincsenek új módosítások	Automatikusan várja meg az értéket, WithPollInterval majd ellenőrizze újra	Ellenőriznie kell az állapotot, és manuálisan újra kell ellenőriznie
Módosítások feldolgozása egy teljes tárolóból	Igen, és automatikusan párhuzamosított több szál és gép között, amelyek ugyanabból a tárolóból származnak	Igen, és manuálisan párhuzamosítva a következő használatával: FeedRange
Csak egyetlen partíciókulcs módosításainak feldolgozása	Nem támogatott	Yes

Megjegyzés

A lekéréses modell használatakor – a változáscsatorna feldolgozójának használatával történő olvasástól eltérően – explicit módon kell kezelnie azokat az eseteket, amikor nincsenek új módosítások.

A lekéréses modell működése

.NET

A változáscsatorna lekéréses modellel történő feldolgozásához hozza létre a példányát.FeedIterator A létrehozásakor FeedIteratormeg kell adnia egy kötelező ChangeFeedStartFrom értéket, amely a módosítások olvasásának kezdőpozícióját és a használni FeedRangekívánt értéket tartalmazza. A FeedRange a partíciókulcs-értékek tartománya, és meghatározza azokat az elemeket, amelyek az adott FeedIteratoradatcsatorna használatával olvashatók a változáscsatornából. Meg kell adnia egy kötelező ChangeFeedMode értéket ahhoz a módhoz is, amelyben a módosításokat szeretné feldolgozni: a legújabb verziót vagy az összes verziót és törlést. ChangeFeedMode.LatestVersion A vagy ChangeFeedMode.AllVersionsAndDeletes a használatával jelezheti, hogy melyik módot szeretné használni a változáscsatorna olvasásához. Ha az összes verziót és törlési módot használja, ki kell választania egy változáscsatorna kezdőértékét vagy Now() egy adott folytatási jogkivonatból.

Igény szerint megadhatja ChangeFeedRequestOptions a beállítását PageSizeHint. Ha be van állítva, ez a tulajdonság beállítja az oldalonként fogadott elemek maximális számát. Ha a figyelt gyűjteményben tárolt eljárásokkal hajtják végre a műveleteket, a tranzakció hatóköre megmarad, amikor elemeket olvas be a változáscsatornából. Ennek eredményeképpen a fogadott elemek száma magasabb lehet a megadott értéknél, így az ugyanazon tranzakció által módosított elemek egy atomi köteg részeként lesznek visszaadva.

Íme egy példa arra, hogyan szerezhető be FeedIterator a legújabb verzió módban, amely entitásobjektumokat ad vissza, ebben az esetben egy User objektumot:

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

Tipp

A verzió 3.34.0előtt a legújabb verziómód használható a beállítással ChangeFeedMode.Incremental. LatestVersion A változáscsatorna legújabb verziómódja és a két módot használó alkalmazások is Incremental ugyanazt a viselkedést fogják tapasztalni.

Minden verzió és törlési mód előzetes verzióban érhető el, és a .NET SDK előzetes verziójával >használható = 3.32.0-preview. Íme egy példa a dinamikus objektumokat visszamutató FeedIterator összes verzió és törlési mód beszerzésére:

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

Megjegyzés

A legújabb verzió módban olyan objektumokat kap, amelyek a módosított elemet jelölik, néhány további metaadattal együtt. Minden verzió és törlési mód egy másik adatmodellt ad vissza. További információ: A válaszobjektum elemzése.

A változáscsatorna felhasználása streameken keresztül

FeedIterator mindkét változáscsatorna módhoz két lehetőség áll rendelkezésre. Az entitásobjektumokat visszaadandó példákon kívül a választ is beszerezheti támogatással Stream . A streamek lehetővé teszik az adatok olvasását anélkül, hogy először deszerializálni kellene őket, így mentheti az ügyfélerőforrásokat.

Íme egy példa arra, hogyan szerezhető be FeedIterator a legújabb verzió módban, amely a következőt adja Streamvissza:

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

Egy teljes tároló módosításainak felhasználása

Ha nem ad meg paramétert FeedRange a számára FeedIterator, a teljes tároló változáscsatornáját a saját tempójában dolgozhatja fel. Íme egy példa, amely elkezdi beolvasni az összes módosítást az aktuális időponttól kezdve a legújabb verziómód használatával:

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

Mivel a változáscsatorna gyakorlatilag az összes jövőbeli írást és frissítést magában foglaló elemek végtelen listája, a értéke HasMoreResults mindig true. Amikor megpróbálja elolvasni a változáscsatornát, és nincsenek elérhető új módosítások, állapotú NotModified választ kap. Az előző példában úgy kezeli a rendszer, hogy öt másodpercet vár a módosítások ismételt ellenőrzése előtt.

Partíciókulcs módosításainak felhasználása

Bizonyos esetekben előfordulhat, hogy csak egy adott partíciókulcs módosításait szeretné feldolgozni. Beszerezhet egy adott partíciókulcsot, és ugyanúgy dolgozhatja FeedIterator fel a módosításokat, mint egy teljes tároló esetében.

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

A FeedRange használata párhuzamosításhoz

A változáscsatorna feldolgozójában a munka automatikusan több felhasználó között oszlik meg. A változáscsatorna lekérési modelljében a FeedRange használatával párhuzamosíthatja a változáscsatorna feldolgozását. Az A FeedRange a partíciókulcs-értékek tartományát jelöli.

Íme egy példa, amely bemutatja, hogyan szerezheti be a tároló tartományainak listáját:

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

Ha lekér egy értéklistát FeedRange a tárolóhoz, fizikai partíciónként egyet FeedRange kap.

A használatával FeedRangelétrehozhat egy parancsot FeedIterator , amellyel párhuzamosíthatja a változáscsatorna feldolgozását több gépen vagy szálon. Az előző példától eltérően, amely azt mutatta be, hogyan szerezhető be a FeedIterator teljes tárolóhoz vagy egyetlen partíciókulcshoz, a FeedRanges használatával több FeedIteratort is beszerezhet, amelyek párhuzamosan tudják feldolgozni a változáscsatornát.

Abban az esetben, ha a FeedRanges-t szeretné használni, rendelkeznie kell egy vezénylőfolyamatkal, amely beszerzi a FeedRanges-eket, és osztja el őket az adott gépeken. Ez a disztribúció a következő lehet:

• A sztringérték használata FeedRange.ToJsonString és terjesztése. A fogyasztók ezt az értéket használhatják a következővel FeedRange.FromJsonString: .
• Ha az eloszlás folyamatban van, adja át az FeedRange objektumhivatkozást.

Íme egy minta, amely bemutatja, hogyan olvashatja be a tároló változáscsatornájának kezdetét két, párhuzamosan olvasható, hipotetikus, különálló gép használatával:

1. gép:

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. gép:

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

Folytatási jogkivonatok mentése

A folytatási jogkivonat beszerzésével mentheti a FeedIterator pozícióját. A folytatási jogkivonat egy sztringérték, amely nyomon követi a FeedIterator utolsó feldolgozott módosításait, és lehetővé teszi a FeedIterator későbbi folytatást. Ha meg van adva a folytatási jogkivonat, elsőbbséget élvez a kezdési időponttal szemben, és az első értékektől indul. A következő kód végigolvassa a változáscsatornát a tároló létrehozása óta. Miután nem érhető el több módosítás, megőrzi a folytatási jogkivonatot, hogy a változáscsatorna-használat később folytatódjon.

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

A legújabb verziómód használata esetén a FeedIterator folytatási jogkivonat soha nem jár le, amíg az Azure Cosmos DB-tároló még létezik. Ha az összes verziót és törlési módot használja, a folytatási FeedIterator jogkivonat akkor érvényes, ha a módosítások a folyamatos biztonsági mentések megőrzési időszakában történtek.

Java

A változáscsatorna lekéréses modellel történő feldolgozásához hozza létre a példányát.Iterator<FeedResponse<JsonNode>> responseIterator A létrehozásakor CosmosChangeFeedRequestOptionsmeg kell adnia, hogy honnan kezdje el olvasni a változáscsatornát, és adja meg a FeedRange használni kívánt paramétert. A FeedRange a partíciókulcs-értékek tartománya, amely meghatározza a változáscsatornából beolvasható elemeket.

Ha az összes verzióban és törlési módban szeretné olvasni a változáscsatornát, meg kell adnia allVersionsAndDeletes() azt is, hogy mikor hozza létre a következőt CosmosChangeFeedRequestOptions: . Az összes verzió és törlési mód nem támogatja a változáscsatorna feldolgozását az elejétől vagy egy időponttól kezdve. A módosításokat mostantól vagy folytatási jogkivonatból kell feldolgoznia. Minden verzió és törlési mód előzetes verzióban érhető el, és elérhető a Java SDK verziójában >= 4.42.0.

Egy teljes tároló módosításainak felhasználása

Ha a értéket adja meg FeedRange.forFullRange(), a saját tempójában feldolgozhatja egy teljes tároló változáscsatornáját. Igény szerint megadhat egy értéket a fájlban byPage(). Ha be van állítva, ez a tulajdonság beállítja az oldalonként fogadott elemek maximális számát.

Megjegyzés

Az alábbi kódrészletek mindegyike egy GitHub-mintából származik. Használhatja a legújabb verziómód-mintát , valamint az összes verziót és törlési módot.

Íme egy példa arra, hogyan szerezhet be responseIterator értéket a legújabb verzió módban:

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

Íme egy példa arra, hogyan szerezhet be egy-egy verziót responseIterator az összes verzióban és törlési módban:

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

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

Ezután iterálhatunk az eredményeken. Mivel a változáscsatorna gyakorlatilag az összes jövőbeli írást és frissítést magában foglaló elemek végtelen listája, a értéke responseIterator.hasNext() mindig true. Íme egy példa a legújabb verzió módban, amely az elejétől kezdve beolvassa az összes módosítást. Minden iteráció megőrzi a folytatási jogkivonatot, miután feldolgozta az összes eseményt. A változáscsatorna utolsó feldolgozott pontjáról veszi fel, és a használatával createForProcessingFromContinuationkezeli:

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

Partíciókulcs módosításainak felhasználása

Bizonyos esetekben előfordulhat, hogy csak egy adott partíciókulcs módosításait szeretné feldolgozni. Egy adott partíciókulcs módosításait ugyanúgy dolgozhatja fel, mint egy teljes tároló esetében. Íme egy példa, amely a legújabb verziómódot használja:

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

A FeedRange használata párhuzamosításhoz

A változáscsatorna feldolgozójában a munka automatikusan több felhasználó között oszlik meg. A változáscsatorna lekérési modelljében a FeedRange használatával párhuzamosíthatja a változáscsatorna feldolgozását. Az A FeedRange a partíciókulcs-értékek tartományát jelöli.

Íme egy példa, amely a legújabb verziómódot használja, amely bemutatja, hogyan szerezheti be a tároló tartományainak listáját:

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

Ha beszerezi a tárolóHoz tartozó FeedRanges-listát, fizikai partíciónként egyet FeedRange kap.

A használatával FeedRangepárhuzamosíthatja a változáscsatorna feldolgozását több gépen vagy szálon. Az előző példától eltérően, amely a teljes tároló vagy egyetlen partíciókulcs módosításainak feldolgozását mutatta be, a FeedRanges használatával párhuzamosan is feldolgozhatja a változáscsatornát.

Abban az esetben, ha a FeedRanges-t szeretné használni, rendelkeznie kell egy vezénylőfolyamatkal, amely beszerzi a FeedRanges-eket, és osztja el őket az adott gépeken. Ez a disztribúció a következő lehet:

• A sztringérték használata FeedRange.toString() és terjesztése.
• Ha az eloszlás folyamatban van, adja át az FeedRange objektumhivatkozást.

Íme egy minta, amely a legújabb verziómódot használja. Bemutatja, hogyan olvashatja be a tároló változáscsatornájának kezdetét két, párhuzamosan olvasható, hipotetikus, különálló gép használatával:

1. gép:

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. gép:

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

A változáscsatorna lekéréses modellel történő feldolgozásához hozza létre a példányát.ChangeFeedPullModelIterator A létrehozásakor ChangeFeedPullModelIteratormeg kell adnia egy kötelező changeFeedStartFrom értéket a ChangeFeedIteratorOptions fájlon belül, amely a módosítások olvasásának kezdőpozícióját és azt az erőforrást (partíciókulcsot vagy FeedRange-t) tartalmazza, amelyhez a módosításokat le kell kérni.

Megjegyzés

Ha nincs changeFeedStartFrom megadva érték, akkor a changefeed egy teljes tárolóhoz lesz lekérve a Now() fájlból. Jelenleg csak a legújabb verziót támogatja a JS SDK, és alapértelmezés szerint ki van választva.

A(z) használatával megadhatja maxItemCountChangeFeedIteratorOptions az egy oldalonként fogadott elemek maximális számát. Íme egy példa arra, hogyan szerezheti be az iterátort a legújabb verziómódban, amely entitásobjektumokat ad vissza:

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

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

Egy teljes tároló módosításainak felhasználása

Ha nem ad meg vagy PartitionKey paramétert FeedRange a fájlbanChangeFeedStartFrom, a saját tempójában dolgozhatja fel egy teljes tároló változáscsatornáját. Íme egy példa, amely elkezdi beolvasni az összes módosítást az aktuális időponttól kezdve:

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

Mivel a változáscsatorna gyakorlatilag az összes jövőbeli írást és frissítést magában foglaló elemek végtelen listája, a értéke hasMoreResults mindig true. Amikor megpróbálja elolvasni a változáscsatornát, és nincsenek elérhető új módosítások, állapotú NotModified választ kap. Az előző példában úgy kezeli a rendszer, hogy öt másodpercet vár a módosítások ismételt ellenőrzése előtt.

Partíciókulcs módosításainak felhasználása

Bizonyos esetekben előfordulhat, hogy csak egy adott partíciókulcs módosításait szeretné feldolgozni. Beszerezheti az iterátort egy adott partíciókulcshoz, és ugyanúgy dolgozhatja fel a módosításokat, mint egy teljes tároló esetében.

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

A FeedRange használata párhuzamosításhoz

A változáscsatorna lekérési modelljében a FeedRange használatával párhuzamosíthatja a változáscsatorna feldolgozását. Az A FeedRange a partíciókulcs-értékek tartományát jelöli.

Íme egy példa, amely bemutatja, hogyan szerezheti be a tároló tartományainak listáját:

const ranges = await container.getFeedRanges();

Ha lekér egy értéklistát FeedRange a tárolóhoz, fizikai partíciónként egyet FeedRange kap.

A használatával FeedRangeiterátort hozhat létre a változáscsatorna több gépen vagy szálon történő feldolgozásának párhuzamosításához. Az előző példától eltérően, amely azt mutatta be, hogyan szerezhet be egy changefeed iteratort a teljes tárolóhoz vagy egyetlen partíciókulcshoz, a FeedRanges használatával több iterátort is beszerezhet, amelyek párhuzamosan tudják feldolgozni a változáscsatornát.

Íme egy minta, amely bemutatja, hogyan olvashatja be a tároló változáscsatornájának kezdetét két, párhuzamosan olvasható, hipotetikus, különálló gép használatával:

1. gép:

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. gép:

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

Folytatási jogkivonatok mentése

A folytatási jogkivonat beszerzésével mentheti az iterátor pozícióját. A folytatási jogkivonat egy sztringérték, amely nyomon követi a changefeed iterator utolsó feldolgozott módosításait, és lehetővé teszi az iterátor későbbi folytatását. Ha meg van adva a folytatási jogkivonat, elsőbbséget élvez a kezdési időponttal szemben, és az első értékektől indul. A következő kód végigolvassa a változáscsatornát a tároló létrehozása óta. Miután nem érhető el több módosítás, megőrzi a folytatási jogkivonatot, hogy a változáscsatorna-használat később folytatódjon.

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

A folytatási jogkivonat soha nem jár le, amíg az Azure Cosmos DB-tároló még létezik.

Az AsyncIterator használata

A Változáscsatorna lekéréséhez használhatja a JavaScript Async Iteratort. Íme egy példa az Async Iterator használatára.

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

Következő lépések

• A változáscsatorna áttekintése
• A változáscsatorna feldolgozójának használata
• Az Azure Functions aktiválása