Use o estimador do fluxo de alterações

Como modelo push para notificações automáticas

Tal como o processador de feed de alterações, o estimador de alterações pode funcionar como um modelo push. O estimador mede a diferença entre o último item processado (definido pelo estado do contentor do arrendamento) e a última alteração no contentor, e envia esse valor para um delegado. O intervalo em que a medição é feita também pode ser personalizado com um valor padrão de 5 segundos.

Por exemplo, se o seu processador de feed de alterações estiver a usar o modo de última versão e estiver definido assim:

Container leaseContainer = client.GetContainer(databaseId, Program.leasesContainer);
Container monitoredContainer = client.GetContainer(databaseId, Program.monitoredContainer);
ChangeFeedProcessor changeFeedProcessor = monitoredContainer
    .GetChangeFeedProcessorBuilder<ToDoItem>("changeFeedEstimator", Program.HandleChangesAsync)
        .WithInstanceName("consoleHost")
        .WithLeaseContainer(leaseContainer)
        .Build();

A forma correta de inicializar um estimador para medir esse processador seria usando GetChangeFeedEstimatorBuilder a seguinte:

ChangeFeedProcessor changeFeedEstimator = monitoredContainer
    .GetChangeFeedEstimatorBuilder("changeFeedEstimator", Program.HandleEstimationAsync, TimeSpan.FromMilliseconds(1000))
    .WithLeaseContainer(leaseContainer)
    .Build();

Onde tanto o processador quanto o estimador partilham o mesmo leaseContainer e o mesmo nome.

Os outros dois parâmetros são o delegado, que recebe um número que representa quantas alterações estão pendentes de serem lidas pelo processador, e o intervalo de tempo em que se pretende que esta medição seja feita.

Um exemplo de delegado que recebe a estimativa é:

static async Task HandleEstimationAsync(long estimation, CancellationToken cancellationToken)
{
    if (estimation > 0)
    {
        Console.WriteLine($"\tEstimator detected {estimation} items pending to be read by the Processor.");
    }

    await Task.Delay(0);
}

Pode enviar esta estimativa para a sua solução de monitorização e usá-la para perceber como o seu progresso está a evoluir ao longo do tempo.

Como uma estimativa detalhada sob demanda

Ao contrário do modelo push, existe uma alternativa que permite obter o orçamento sob demanda. Este modelo também fornece informações mais detalhadas:

• O atraso estimado por concessão.
• A instância que detém e processa cada arrendamento, para que possas identificar se há algum problema numa instância.

Se o seu processador de alimentação de alterações estiver definido assim:

Container leaseContainer = client.GetContainer(databaseId, Program.leasesContainer);
Container monitoredContainer = client.GetContainer(databaseId, Program.monitoredContainer);
ChangeFeedProcessor changeFeedProcessor = monitoredContainer
    .GetChangeFeedProcessorBuilder<ToDoItem>("changeFeedEstimator", Program.HandleChangesAsync)
        .WithInstanceName("consoleHost")
        .WithLeaseContainer(leaseContainer)
        .Build();

Pode criar o estimador com a mesma configuração de leasing:

ChangeFeedEstimator changeFeedEstimator = monitoredContainer
    .GetChangeFeedEstimator("changeFeedEstimator", leaseContainer);

E sempre que quiser, com a frequência que precisa, pode obter a estimativa detalhada:

Console.WriteLine("Checking estimation...");
using FeedIterator<ChangeFeedProcessorState> estimatorIterator = changeFeedEstimator.GetCurrentStateIterator();
while (estimatorIterator.HasMoreResults)
{
    FeedResponse<ChangeFeedProcessorState> states = await estimatorIterator.ReadNextAsync();
    foreach (ChangeFeedProcessorState leaseState in states)
    {
        string host = leaseState.InstanceName == null ? $"not owned by any host currently" : $"owned by host {leaseState.InstanceName}";
        Console.WriteLine($"Lease [{leaseState.LeaseToken}] {host} reports {leaseState.EstimatedLag} as estimated lag.");
    }
}

Cada ChangeFeedProcessorState contém a informação do arrendamento e do atraso, bem como quem é a instância atual que o detém.

Implantação do estimador

O estimador de feed de alterações não precisa de ser implementado como parte do seu processador de feed de alterações, nem de fazer parte do mesmo projeto. Recomendamos que implemente o estimador numa instância independente dos seus processadores. Uma única instância de estimador pode acompanhar o progresso de todas as locações e instâncias na implementação do seu processador de feed de alterações.

Cada estimativa consome unidades de pedido de cada um dos seus contentores monitorizados e de arrendamento. Uma frequência de 1 minuto de intervalo é um bom ponto de partida; quanto menor a frequência, maiores são as unidades de pedido consumidas.

Este exemplo utiliza o processador de feed de alterações no modo da versão mais recente, com a estimativa do atraso no processamento dos eventos do feed de alterações. Na aplicação – os documentos são inseridos num recipiente (o "recipiente de alimentação"), e entretanto outro fio de trabalho ou aplicação de trabalho está a puxar documentos inseridos do feed de alterações do contentor de alimentação e a operar sobre eles de alguma forma.

O Change Feed Processor é implementado e iniciado da seguinte maneira:

ChangeFeedProcessor changeFeedProcessorMainInstance = new ChangeFeedProcessorBuilder()
    .hostName("SampleHost_1")
    .feedContainer(feedContainer)
    .leaseContainer(leaseContainer)
    .handleChanges(handleChangesWithLag())
    .buildChangeFeedProcessor();

try {
    changeFeedProcessorMainInstance
        .start()
        .subscribeOn(Schedulers.boundedElastic())
        .timeout(Duration.ofSeconds(10))
        .then(Mono.just(changeFeedProcessorMainInstance)
                  .delayElement(Duration.ofSeconds(10))
                  .flatMap(value -> changeFeedProcessorMainInstance
                      .stop()
                      .subscribeOn(Schedulers.boundedElastic())
                      .timeout(Duration.ofSeconds(10))))
        .subscribe();
} catch (Exception ex) {
    logger.error("Change feed processor did not start and stopped in the expected time", ex);
    throw ex;
}

A verificação do lag do processador de feed de alterações pode ser realizada numa aplicação separada (como um monitor de saúde) desde que sejam usados os mesmos contentores de entrada (feedContainer e leaseContainer) e o mesmo prefixo de arrendamento (CONTAINER_NAME + "-lease"). O código do estimador exige que o processador do feed de alterações tenha tido a oportunidade de inicializar totalmente os documentos do leaseContainer.

O estimador calcula o atraso acumulado recuperando o estado atual do processador de alimentação de alterações e somando os valores estimados de atraso para cada evento a ser processado:

Mono<List<ChangeFeedProcessorState>> currentState = changeFeedProcessorMainInstance.getCurrentState();
currentState.map(state -> {
    for (ChangeFeedProcessorState changeFeedProcessorState : state) {
        totalLag.addAndGet(changeFeedProcessorState.getEstimatedLag());
    }
    return state;
}).subscribe();

// Initially totalLag should be zero
logger.info("Initially total lag is : {}", totalLag.get());

totalLag.set(0);
currentState = changeFeedProcessorSideCart.getCurrentState();
currentState.map(state -> {
    for (ChangeFeedProcessorState changeFeedProcessorState : state) {
        totalLag.addAndGet(changeFeedProcessorState.getEstimatedLag());
    }
    return state;
}).subscribe();

// Initially totalLag should be zero
logger.info("Initially total lag is : {}", totalLag.get());

O atraso total deve inicialmente ser zero e, finalmente, deve ser maior ou igual ao número de documentos criados. O valor total do atraso pode ser registado ou usado para análises adicionais, permitindo monitorizar o desempenho do processamento do feed de alterações e identificar eventuais gargalos ou atrasos no sistema:

totalLag.set(0);
changeFeedProcessorSideCart.start().block();
currentState = changeFeedProcessorSideCart.getCurrentState();
currentState.map(state -> {
    for (ChangeFeedProcessorState changeFeedProcessorState : state) {
        totalLag.addAndGet(changeFeedProcessorState.getEstimatedLag());
    }
    return state;
}).block();

// Finally, totalLag should be greater or equal to the number of documents created
logger.info("Finally total lag is : {}", totalLag.get());

Um exemplo de delegado que recebe alterações e as trata com atraso é:

private static Consumer<List<JsonNode>> handleChangesWithLag() {
    return (List<JsonNode> docs) -> {
        logger.info("Start handleChangesWithLag()");

        try {
            Thread.sleep(Duration.ofSeconds(5).toMillis());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        for (JsonNode document : docs) {
            try {
                //Change Feed hands the document to you in the form of a JsonNode
                //As a developer you have two options for handling the JsonNode document provided to you by Change Feed
                //One option is to operate on the document in the form of a JsonNode, as shown below. This is great
                //especially if you do not have a single uniform data model for all documents.
                logger.info("Document received: " + OBJECT_MAPPER.writerWithDefaultPrettyPrinter()
                                                                 .writeValueAsString(document));

                //You can also transform the JsonNode to a POJO having the same structure as the JsonNode,
                //as shown below. Then you can operate on the POJO.
                CustomPOJO2 pojo_doc = OBJECT_MAPPER.treeToValue(document, CustomPOJO2.class);
                logger.info("id: " + pojo_doc.getId());

            } catch (JsonProcessingException e) {
                e.printStackTrace();
            }
        }
        logger.info("End handleChangesWithLag()");
    };
}