Alterar o modelo de pull do feed de alterações no Azure Cosmos DB

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

Opcionalmente, você pode especificar ChangeFeedRequestOptions para definir um PageSizeHint. 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 durante a leitura de itens do feed de alterações. Como resultado, o número de itens recebidos pode ser maior 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 para obter um FeedIterator no modo de versão mais recente que retorna objetos de entidade, neste caso, um objeto User:

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

O modo de todas as versões e exclusões está em versão prévia e pode ser usado com a versão >= 3.32.0-preview do SDK do .NET. Aqui está um exemplo para obter FeedIterator em todas as versões e excluir o modo que retorna objetos User:

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

Consumir o feed de alterações por meio de fluxos

FeedIterator para ambos os modos de feed de alterações existem duas opções. Além dos exemplos que retornam objetos de entidade, você também pode obter a resposta com suporte a Stream. Os fluxos permitem que você leia os dados sem necessidade de desserializá-los primeiro, economizando recursos do cliente.

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

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

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

Se você não fornecer um parâmetro FeedRange para FeedIterator, poderá processar o feed de alterações de um contêiner inteiro em 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 abrangem todas as futuras gravações e atualizações, o valor de HasMoreResults é sempre true. Ao tentar ler o feed de alterações e não houver nenhuma nova alteração disponível, você recebe uma resposta com o status NotModified. No exemplo acima, ela é tratada aguardando cinco segundos antes de verificar novamente se há alterações.

Consumir as alterações para 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 para uma chave de partição específica e processar as alterações da mesma maneira que pode fazê-lo 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 do feed de alterações, o trabalho é distribuído automaticamente entre vários consumidores. No modelo de pull do feed de alterações, você pode usar o FeedRange para paralelizar o processamento do feed de alterações. Um FeedRange representa um intervalo de valores de chave de partição.

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

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

Ao obter a lista de valores FeedRange para seu contêiner, você 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ários computadores ou threads. Ao contrário do exemplo anterior, que mostrou como obter um FeedIterator para o contêiner inteiro ou uma única chave de partição, é possível usar FeedRanges para obter vários FeedIterators que podem processar o feed de alterações em paralelo.

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

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

Aqui está uma amostra que mostra como ler desde o início do feed de alterações do contêiner usando dois computadores hipotéticos separados que estão realizando a leitura 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 de 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 FeedIterator e permite que FeedIterator retome nesse ponto mais tarde. O token de continuação, se especificado, tem precedência sobre os valores da Hora de início e Começar do início. 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 manterá um token de continuação para que o consumo do 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 continuação FeedIterator nunca expirará enquanto o contêiner do Azure Cosmos DB ainda existir. Quando você estiver usando o modo todas as versões e exclusões, o token de continuação FeedIterator é 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 de pull, crie uma instância de Iterator<FeedResponse<JsonNode>> responseIterator. Ao criar CosmosChangeFeedRequestOptions, você deve especificar de onde começar a ler o feed de alterações e passar o parâmetro FeedRange que deseja usar. O FeedRange é um intervalo de valores de chave de partição que especifica os itens que podem ser lidos do feed de alterações.

Se você quiser ler o feed de alterações no modo de todas as versões e exclusões, especifique allVersionsAndDeletes() ao criar o CosmosChangeFeedRequestOptions. O modo de todas as versões e exclusões não dá suporte ao processamento do feed de alterações desde o início ou de um momento específico. Você deve processar alterações a partir de agora ou de um token de continuação. O modo de todas as versões e exclusões está em versão prévia e disponível na versão do SDK do Java > = 4.42.0.

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

Se você especificar FeedRange.forFullRange(), poderá processar o feed de alterações de um contêiner inteiro em 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 para obter um valor responseIterator 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 no modo de todas as versões e exclusões:

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

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

Em seguida, podemos iterar sobre os resultados. Como o feed de alterações é efetivamente uma lista infinita de itens que abrangem todas as futuras gravações e atualizações, o valor de responseIterator.hasNext() é sempre true. Aqui está um exemplo no modo de versão mais recente, que lê todas as alterações a partir do início. Cada iteração persiste um token de continuação depois de processar todos os eventos. Ele pega do último ponto processado no feed de alterações 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 maneira que pode fazê-lo para um contêiner inteiro. Aqui está um exemplo usando 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 do feed de alterações, o trabalho é distribuído automaticamente entre vários consumidores. No modelo de pull do feed de alterações, você pode usar o FeedRange para paralelizar o processamento do feed de alterações. Um FeedRange representa um intervalo de valores de chave de partição.

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

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

Ao obter a lista de FeedRanges para seu contêiner, você obtém um FeedRange por partição física.

Usando um FeedRange, você pode paralelizar o processamento do feed de alterações em vários computadores ou threads. Diferente do exemplo anterior, que mostrou como processar alterações para o contêiner inteiro ou uma só chave de partição, é possível usar FeedRanges para processar o feed de alterações em paralelo.

No caso em que você deseja usar o FeedRanges, você precisa ter um processo orquestrador que obtém FeedRanges e os distribui para esses computadores. Essa 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 objeto FeedRange.

Aqui está uma amostra usando o modo de versão mais recente. Ela mostra como ler desde o início do feed de alterações do contêiner usando dois computadores hipotéticos separados que estão realizando a leitura 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 de pull, crie uma instância de ChangeFeedPullModelIterator. Ao criar ChangeFeedPullModelIterator inicialmente, você deve especificar um valor changeFeedStartFrom necessário dentro do ChangeFeedIteratorOptions que consiste na posição inicial para ler as alterações e o recurso (uma chave de partição ou um FeedRange) para o qual as alterações devem ser buscadas.

Opcionalmente, você pode usar maxItemCount em ChangeFeedIteratorOptions para 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 para um contêiner inteiro

Se você não fornecer um parâmetro FeedRange ou PartitionKey dentro de ChangeFeedStartFrom, você poderá processar o feed de alterações de um contêiner inteiro em 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 abrangem todas as futuras gravações e atualizações, o valor de hasMoreResults é sempre true. Ao tentar ler o feed de alterações e não houver nenhuma nova alteração disponível, você recebe uma resposta com o status NotModified. No exemplo acima, ela é tratada aguardando cinco segundos antes de verificar novamente se há alterações.

Consumir as alterações para 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 um iterador para uma chave de partição específica e processar as alterações da mesma maneira que pode fazê-lo 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 do feed de alterações, você pode usar o FeedRange para paralelizar o processamento do feed de alterações. Um FeedRange representa um intervalo de valores de chave de partição.

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

const ranges = await container.getFeedRanges();

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

Ao usar um FeedRange, você pode criar um iterador para paralelizar o processamento do feed de alterações em vários computadores ou threads. Ao contrário do exemplo anterior, que mostrou como obter um iterador de changefeed para o contêiner inteiro ou uma única chave de partição, é possível usar FeedRanges para obter vários iteradores que podem processar o feed de alterações em paralelo.

Aqui está uma amostra que mostra como ler desde o início do feed de alterações do contêiner usando dois computadores hipotéticos separados que estão realizando a leitura 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 de seu 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 retome nesse ponto mais tarde. O token de continuação, se especificado, tem precedência sobre os valores da Hora de início e Começar do início. 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 manterá um token de continuação para que o consumo do 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 contentor DB do Azure Cosmos ainda existir.

Usar AsyncIterator

Você pode usar o Iterador Assíncrono do JavaScript para buscar o feed de alterações. Aqui está um exemplo para usar o Iterador Assíncrono.

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