Utilice el estimador de flujo de cambios

Como modelo de empuje para notificaciones automáticas

Al igual que el procesador de fuente de cambios, el estimador de fuente de cambios puede operar como modelo de inserción. El estimador mide la diferencia entre el último elemento procesado (definido por el estado del contenedor de arrendamientos) y el último cambio en el contenedor, y envía este valor a un delegado. El intervalo en el que se toma la medida también se puede personalizar con un valor predeterminado de 5 segundos.

Por ejemplo, si el procesador de cambio de datos utiliza el modo de versión más reciente y está definido de esta forma:

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

La manera correcta de inicializar un estimador para medir ese procesador estaría usando GetChangeFeedEstimatorBuilder de la siguiente manera:

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

Donde tanto el procesador como el estimador comparten el mismo leaseContainer y el mismo nombre.

Los otros dos parámetros son el delegado, que recibe un número que representa cuántos cambios están pendientes de ser leídos por el procesador y el intervalo de tiempo en el que desea que se tome esta medida.

Un ejemplo de delegado que recibe la estimación es:

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

Puede enviar esta estimación a la solución de supervisión y usarla para comprender cómo se comporta el progreso con el tiempo.

Como estimación detallada a petición

A diferencia del modelo push, hay una alternativa que le permite obtener la estimación bajo demanda. Este modelo también proporciona información más detallada:

• Retraso estimado por concesión.
• La instancia propietaria y el procesamiento de cada concesión, por lo que puede identificar si hay un problema en una instancia.

Si el procesador de flujo de cambios se define de la siguiente manera:

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

Puede crear el estimador con la misma configuración de concesión:

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

Y siempre que quiera, con la frecuencia que necesite, puede obtener la estimación detallada:

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 contiene la información de arrendamiento y retraso, y también cuál es la instancia actual que lo posee.

Implementación del estimador

No es necesario implementar el estimador de fuente de cambios como parte del procesador de fuente de cambios ni formar parte del mismo proyecto. Se recomienda implementar el estimador en una instancia independiente de los procesadores. Una sola instancia de estimador puede realizar un seguimiento del progreso de todos los arrendamientos e instancias en la implementación del procesador de feed de cambios.

Cada estimación consume unidades de solicitud de contenedores supervisados y arrendados. Una frecuencia de 1 minuto entre es un buen punto de partida, cuanto menor sea la frecuencia, mayor será el consumo de las unidades de solicitud.

En este ejemplo se utiliza el procesador de 'change feed' en el modo de última versión, estimando el retraso en el procesamiento de eventos del 'change feed'. En la aplicación: los documentos se insertan en un contenedor (el "contenedor de alimentación"), y, mientras tanto, otra aplicación de trabajo o subproceso de trabajo extrae documentos insertados del feed de cambios del contenedor de alimentación y realiza operaciones sobre ellos de alguna manera.

El procesador de alimentación de cambios se construye e inicia de la siguiente manera:

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

La comprobación del retardo del procesador de cambios puede realizarse en una aplicación independiente (como un monitor de salud) siempre que se usen los mismos contenedores de entrada (feedContainer y leaseContainer) y el mismo prefijo de concesión (CONTAINER_NAME + "-lease"). El código del estimador requiere que el procesador de fuente de cambios tenga la oportunidad de inicializar completamente los documentos del contenedor de arrendamiento.

El estimador calcula el retraso acumulado recuperando el estado actual del procesador de fuente de cambios y sumando los valores de retraso estimados para cada evento que se está procesando:

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

El retraso total inicialmente debe ser cero y, por último, debe ser mayor o igual que el número de documentos creados. El valor total de retraso se puede registrar o usar para un análisis posterior, lo que permite supervisar el rendimiento del procesamiento de la fuente de cambios e identificar posibles cuellos de botella o retrasos en el 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());

Un ejemplo de delegado que recibe los cambios y los controla con un retraso es:

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