Alterar o modelo de pull de feed no Azure Cosmos DB

Para processar o feed de alterações usando o modelo pull, crie uma instância de FeedIterator. Ao criar FeedIteratorinicialmente o , você deve especificar um valor necessário ChangeFeedStartFrom , que consiste na posição inicial para ler as alterações e no valor que deseja usar para FeedRange. O FeedRange é um intervalo de valores de chave de partição e especifica os itens que podem ser lidos a partir do feed de alterações usando esse arquivo .FeedIterator Você também deve especificar um valor necessário ChangeFeedMode para o modo no qual deseja processar as alterações: versão mais recente ou todas as versões e exclusões. Use um ou ChangeFeedMode.LatestVersion ChangeFeedMode.AllVersionsAndDeletes para indicar qual modo você deseja usar para ler o feed de alterações. Ao usar todas as versões e o modo de exclusão, você deve selecionar um início de feed de alteração a partir do valor de um Now() ou de um token de continuação específico.

Opcionalmente, você pode especificar ChangeFeedRequestOptions para definir um PageSizeHintarquivo . Quando definida, essa propriedade define o número máximo de itens recebidos por página. Se as operações na coleção monitorada forem executadas por meio de procedimentos armazenados, o escopo da transação será preservado ao ler itens do feed de alterações. Como resultado, o número de itens recebidos pode ser maior do que o valor especificado para que os itens alterados pela mesma transação sejam retornados como parte de um lote atômico.

Aqui está um exemplo de como obter FeedIterator no modo de versão mais recente que retorna objetos de entidade, neste caso, um User objeto:

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

Todas as versões e exclusões do modo está em visualização e pode ser usado com versões de visualização do >SDK do .NET = 3.32.0-preview. Aqui está um exemplo para obter FeedIterator em todas as versões e exclui o modo que retorna User objetos:

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

Consuma o feed de alterações através de streams

FeedIterator para ambos os modos de alimentação de mudança tem duas opções. Além dos exemplos que retornam objetos de entidade, você também pode obter a resposta com Stream suporte. Os fluxos permitem que você leia dados sem tê-los primeiro desserializados, para que você economize nos recursos do cliente.

Aqui está um exemplo de como obter FeedIterator no modo de versão mais recente que retorna Stream:

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

Consumir as alterações de um contêiner inteiro

Se você não fornecer um FeedRange parâmetro para o FeedIterator, poderá processar a alimentação de alteração de um contêiner inteiro no seu próprio ritmo. Aqui está um exemplo, que começa a ler todas as alterações, começando no momento atual usando o modo de versão mais recente:

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

Como o feed de alterações é efetivamente uma lista infinita de itens que englobam todas as gravações e atualizações futuras, o valor de HasMoreResults é sempre true. Quando você tenta ler o feed de alterações e não há novas alterações disponíveis, você recebe uma resposta com NotModified status. No exemplo anterior, ele é tratado aguardando cinco segundos antes de verificar novamente se há alterações.

Consumir as alterações de uma chave de partição

Em alguns casos, talvez você queira processar apenas as alterações para uma chave de partição específica. Você pode obter FeedIterator uma chave de partição específica e processar as alterações da mesma maneira que pode para um contêiner inteiro.

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

Usar FeedRange para paralelização

No processador de alimentação de alterações, o trabalho é automaticamente distribuído por vários consumidores. No modelo de pull de feed de alterações, você pode usar o FeedRange para paralelizar o processamento do feed de alterações. A FeedRange representa um intervalo de valores de chave de partição.

Aqui está um exemplo que mostra como obter uma lista de intervalos para seu contêiner:

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

Quando você obtém uma lista de FeedRange valores para seu contêiner, obtém um FeedRange por partição física.

Usando um FeedRange, você pode criar um FeedIterator para paralelizar o processamento do feed de alterações em várias máquinas ou threads. Ao contrário do exemplo anterior que mostrou como obter um FeedIterator para todo o contêiner ou uma única chave de partição, você pode usar FeedRanges para obter vários FeedIterators, que podem processar o feed de alteração em paralelo.

No caso em que você deseja usar FeedRanges, você precisa ter um processo orquestrador que obtém FeedRanges e os distribui para essas máquinas. Esta distribuição pode ser:

• Usando FeedRange.ToJsonString e distribuindo esse valor de cadeia de caracteres. Os consumidores podem usar este valor com FeedRange.FromJsonString.
• Se a distribuição estiver em processo, passando a referência do FeedRange objeto.

Aqui está um exemplo que mostra como ler desde o início da alimentação de mudança do contêiner usando duas máquinas hipotéticas separadas que leem em paralelo:

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

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

Salvar tokens de continuação

Você pode salvar a posição do seu FeedIterator obtendo o token de continuação. Um token de continuação é um valor de cadeia de caracteres que mantém o controle das últimas alterações processadas do seu FeedIterator e permite que o FeedIterator retome neste ponto mais tarde. O token de continuação, se especificado, tem precedência sobre a hora de início e começa a partir dos valores iniciais. O código a seguir lê o feed de alterações desde a criação do contêiner. Depois que não houver mais alterações disponíveis, ele persistirá um token de continuação para que o consumo de feed de alterações possa ser retomado posteriormente.

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

Quando você estiver usando o modo de versão mais recente, o token de FeedIterator continuação nunca expirará enquanto o contêiner do Azure Cosmos DB ainda existir. Quando você estiver usando todas as versões e o modo de exclusão, o FeedIterator token de continuação será válido desde que as alterações tenham ocorrido dentro da janela de retenção para backups contínuos.

Para processar o feed de alterações usando o modelo pull, crie uma instância de Iterator<FeedResponse<JsonNode>> responseIterator. Ao criar CosmosChangeFeedRequestOptionso , você deve especificar de onde começar a ler o feed de alterações e passar o FeedRange parâmetro que deseja usar. O FeedRange é um intervalo de valores de chave de partição que especifica os itens que podem ser lidos a partir do feed de alterações.

Se quiser ler o feed de alterações em todas as versões e no modo de exclusão, você também deve especificar allVersionsAndDeletes() quando criar o CosmosChangeFeedRequestOptions. Todas as versões e o modo de exclusão não suportam o processamento do feed de alterações desde o início ou a partir de um ponto no tempo. Você deve processar alterações a partir de agora ou de um token de continuação. Todas as versões e o modo de exclusão estão em visualização e estão disponíveis na versão >do Java SDK = 4.42.0.

Consumir as alterações de um contêiner inteiro

Se você especificar FeedRange.forFullRange(), poderá processar o feed de alterações para um contêiner inteiro no seu próprio ritmo. Opcionalmente, você pode especificar um valor em byPage(). Quando definida, essa propriedade define o número máximo de itens recebidos por página.

Aqui está um exemplo de como obter um responseIterator valor no modo de versão mais recente:

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

Aqui está um exemplo de como obter um responseIterator em todas as versões e exclusões modo:

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

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

Podemos então iterar sobre os resultados. Como o feed de alterações é efetivamente uma lista infinita de itens que engloba todas as gravações e atualizações futuras, o valor de responseIterator.hasNext() é sempre true. Aqui está um exemplo no modo de versão mais recente, que lê todas as alterações, começando desde o início. Cada iteração persiste um token de continuação depois de processar todos os eventos. Ele pega a partir do último ponto processado no feed de mudança e é manipulado usando 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;
    }
}

Consumir as alterações de uma chave de partição

Em alguns casos, talvez você queira processar apenas as alterações para uma chave de partição específica. Você pode processar as alterações para uma chave de partição específica da mesma forma que para um contêiner inteiro. Aqui está um exemplo que usa o modo de versão mais recente:

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

Usar FeedRange para paralelização

No processador de alimentação de alterações, o trabalho é automaticamente distribuído por vários consumidores. No modelo de pull de feed de alterações, você pode usar o FeedRange para paralelizar o processamento do feed de alterações. A FeedRange representa um intervalo de valores de chave de partição.

Aqui está um exemplo que usa o modo de versão mais recente mostrando como obter uma lista de intervalos para seu contêiner:

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

Quando você obtém uma lista de FeedRanges para seu contêiner, obtém uma FeedRange por partição física.

Usando um FeedRange, você pode paralelizar o processamento do feed de alterações em várias máquinas ou threads. Ao contrário do exemplo anterior que mostrava como processar alterações para todo o contêiner ou uma única chave de partição, você pode usar FeedRanges para processar o feed de alterações em paralelo.

No caso em que você deseja usar FeedRanges, você precisa ter um processo orquestrador que obtém FeedRanges e os distribui para essas máquinas. Esta distribuição pode ser:

• Usando FeedRange.toString() e distribuindo esse valor de cadeia de caracteres.
• Se a distribuição estiver em processo, passando a referência do FeedRange objeto.

Aqui está um exemplo que usa o modo de versão mais recente. Ele mostra como ler desde o início da alimentação de mudança do contêiner usando duas máquinas separadas hipotéticas que leem em paralelo:

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

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

Para processar o feed de alterações usando o modelo pull, crie uma instância de ChangeFeedPullModelIterator. Ao criar ChangeFeedPullModelIteratorinicialmente o , você deve especificar um valor necessário changeFeedStartFrom dentro do que consiste na posição inicial para ler as alterações e no recurso (uma chave de partição ou um FeedRange) para o ChangeFeedIteratorOptions qual as alterações devem ser buscadas.

Opcionalmente, você pode usar maxItemCount para ChangeFeedIteratorOptions definir o número máximo de itens recebidos por página. Aqui está um exemplo de como obter o iterador no modo de versão mais recente que retorna objetos de entidade:

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

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

Consumir as alterações de um contêiner inteiro

Se você não fornecer um FeedRange parâmetro ou PartitionKey dentro ChangeFeedStartFromdo , poderá processar a alimentação de mudança de um contêiner inteiro no seu próprio ritmo. Aqui está um exemplo, que começa a ler todas as alterações, começando no momento atual:

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

Como o feed de alterações é efetivamente uma lista infinita de itens que englobam todas as gravações e atualizações futuras, o valor de hasMoreResults é sempre true. Quando você tenta ler o feed de alterações e não há novas alterações disponíveis, você recebe uma resposta com NotModified status. No exemplo anterior, ele é tratado aguardando cinco segundos antes de verificar novamente se há alterações.

Consumir as alterações de uma chave de partição

Em alguns casos, talvez você queira processar apenas as alterações para uma chave de partição específica. Você pode obter iterador para uma chave de partição específica e processar as alterações da mesma maneira que pode para um contêiner inteiro.

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

Usar FeedRange para paralelização

No modelo de pull de feed de alterações, você pode usar o FeedRange para paralelizar o processamento do feed de alterações. A FeedRange representa um intervalo de valores de chave de partição.

Aqui está um exemplo que mostra como obter uma lista de intervalos para seu contêiner:

const ranges = await container.getFeedRanges();

Quando você obtém uma lista de FeedRange valores para seu contêiner, obtém um FeedRange por partição física.

Usando um , você pode criar iterador FeedRangepara paralelizar o processamento do feed de alterações em várias máquinas ou threads. Ao contrário do exemplo anterior que mostrou como obter um iterador changefeed para todo o contêiner ou uma única chave de partição, você pode usar FeedRanges para obter vários iteradores, que podem processar o feed de alterações em paralelo.

Aqui está um exemplo que mostra como ler desde o início da alimentação de mudança do contêiner usando duas máquinas hipotéticas separadas que leem em paralelo:

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

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

Salvar tokens de continuação

Você pode salvar a posição do iterador obtendo o token de continuação. Um token de continuação é um valor de cadeia de caracteres que mantém o controle das últimas alterações processadas do iterador do changefeed e permite que o iterador seja retomado neste ponto mais tarde. O token de continuação, se especificado, tem precedência sobre a hora de início e começa a partir dos valores iniciais. O código a seguir lê o feed de alterações desde a criação do contêiner. Depois que não houver mais alterações disponíveis, ele persistirá um token de continuação para que o consumo de feed de alterações possa ser retomado posteriormente.

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

O token de continuação nunca expira enquanto o contêiner do Azure Cosmos DB ainda existir.

Usar AsyncIterator

Você pode usar o JavaScript Async Iterator para buscar o changefeed. Aqui está um exemplo para usar o Async Iterator.

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