Consume data from a tunnel - Tablestore SDK for Java - Tablestore - Centro de documentación de Alibaba Cloud

Tablestore SDK for Java permite consumir datos de forma continua desde un túnel, procesar cada lote mediante una función de callback y configurar heartbeats, checkpoints, pools de hilos y la concurrencia de consumo.

Notas de uso

El periodo de retención de los registros incrementales coincide con el periodo de caducidad de los registros de Stream de la tabla y puede ser de hasta siete días. En un túnel BaseAndStream, si el consumo de datos completos no finaliza dentro de este periodo, se devuelve el error OTSTunnelExpired al iniciar el consumo de datos incrementales. El túnel no podrá continuar consumiendo datos incrementales.
Si el consumo incremental se retrasa más allá del periodo de retención, el túnel podría reanudarse desde los datos disponibles más recientes. Como consecuencia, es posible que algunos datos no se consuman.
Un túnel caducado puede deshabilitarse. Si permanece deshabilitado durante más de 30 días, se elimina y no es posible restaurarlo.

Requisitos previos

Instale Tablestore SDK for Java y initialize TunnelClient.

Descripción de la función

TunnelWorker se conecta a un túnel mediante su ID, utiliza heartbeats para obtener los canales asignados al cliente actual, extrae datos de forma continua y pasa cada lote de registros a IChannelProcessor. Si varias instancias de TunnelWorker consumen el mismo túnel, el servidor distribuye los canales entre los clientes.

Para consumir datos desde un túnel:

Implemente IChannelProcessor. Utilice el método process para procesar cada lote y el método shutdown para liberar los recursos que usa la función de callback.
Cree un objeto TunnelWorkerConfig para configurar la función de callback y el comportamiento de consumo.
Cree un TunnelWorker con el ID del túnel, TunnelClient y TunnelWorkerConfig.

Llame a connectAndWorking para iniciar el consumo.

void process(ProcessRecordsInput input);
void shutdown();

En el siguiente ejemplo se imprime cada registro extraído del túnel y luego se inicia el consumo.

private static class SimpleProcessor implements IChannelProcessor {
    @Override
    public void process(ProcessRecordsInput input) {
        for (StreamRecord record : input.getRecords()) {
            System.out.println(record);
        }
    }

    @Override
    public void shutdown() {
        // Release resources used by the callback.
    }
}

String tunnelId = "example_tunnel_id";
TunnelWorkerConfig config =
        new TunnelWorkerConfig(new SimpleProcessor());
TunnelWorker worker =
        new TunnelWorker(tunnelId, tunnelClient, config);
worker.connectAndWorking();

Importante

connectAndWorking retorna después de iniciar las tareas de consumo en segundo plano. Mantenga el proceso de la aplicación en ejecución. Para detener el consumo, llame a worker.shutdown(), config.shutdown() y tunnelClient.shutdown() en ese orden. worker.shutdown() cierra las conexiones del túnel e invoca el método shutdown de la función de callback. config.shutdown() detiene los pools de hilos de lectura, procesamiento y auxiliares. TunnelWorker registra un hook de apagado de JVM que intenta detener el worker, pero la aplicación debe liberar explícitamente estos recursos.

Parámetros

Worker

El constructor de TunnelWorker incluye los siguientes parámetros.

Nombre	Tipo	Descripción
tunnelId (obligatorio)	String	El ID del túnel. Obténgalo al crear, listar o consultar túneles.
client (obligatorio)	TunnelClientInterface	El `TunnelClient` inicializado.
workerConfig (obligatorio)	TunnelWorkerConfig	La configuración de la función de callback y del comportamiento de consumo.

Configuración de consumo

workerConfig es de tipo TunnelWorkerConfig y contiene los siguientes parámetros.

Nombre	Tipo	Descripción
channelProcessor (obligatorio)	IChannelProcessor	La función de callback para el procesamiento de datos. Este parámetro es obligatorio cuando se utiliza el constructor de `TunnelWorker` con tres parámetros.
heartbeatTimeoutInSec (opcional)	long	El tiempo de espera del heartbeat en segundos. El valor predeterminado es `300` y debe ser mayor que `heartbeatIntervalInSec`. Cuando un heartbeat expira, el servidor considera que el cliente no está disponible y este se reconecta al túnel.
heartbeatIntervalInSec (opcional)	long	El intervalo de heartbeat en segundos. El valor predeterminado es `30` y el mínimo es `5`. Los heartbeats obtienen los canales activos, actualizan sus estados e inicializan las tareas de procesamiento de datos. Este intervalo también afecta el tiempo de calentamiento de `TunnelWorker`.
checkpointIntervalInMillis (opcional)	long	El intervalo con el que se registran los checkpoints de consumo en el servidor, en milisegundos. El valor predeterminado es `5000`. Tunnel Service entrega cada registro al menos una vez y conserva el orden de los registros. Una tarea reiniciada se reanuda desde el último checkpoint, por lo que podría procesar algunos datos más de una vez. Un intervalo más corto reduce el procesamiento duplicado, pero registrar checkpoints con demasiada frecuencia puede disminuir el throughput.
clientTag (opcional)	String	Una etiqueta personalizada del cliente que se utiliza para generar un ID de cliente y distinguir las instancias de `TunnelWorker`. El valor predeterminado es la propiedad del sistema Java `os.name`.
readRecordsExecutor (opcional)	ThreadPoolExecutor	El pool de hilos que extrae los datos. El pool predeterminado tiene `32` hilos principales, hasta `1000` hilos, una capacidad de cola de `16` y un tiempo de keep-alive de `60` segundos.
processRecordsExecutor (opcional)	ThreadPoolExecutor	El pool de hilos que procesa los datos. Su configuración predeterminada es idéntica a la de `readRecordsExecutor`. Para un pool personalizado, configure el número de hilos según la cantidad de canales del túnel.
maxChannelParallel (opcional)	int	El número máximo de canales desde los cuales se extraen y procesan datos simultáneamente. Utilice este parámetro para limitar el uso de memoria. El valor predeterminado es `-1`, que indica que no hay límite. Tablestore SDK for Java 5.10.0 y versiones posteriores admiten este parámetro.
channelHelperExecutor (opcional)	ThreadPoolExecutor	El pool de hilos auxiliar que inicializa canales, programa pipelines y gestiona errores en tiempo de ejecución. Si no se establece este parámetro, se utiliza un pool de hilos en caché.
maxRetryIntervalInMillis (opcional)	int	El intervalo base máximo para el backoff exponencial durante la extracción de datos incrementales, en milisegundos. El valor predeterminado es `2000` y el mínimo es `200`. Si un lote contiene como máximo 500 registros y no supera los 900 KB, el cliente aumenta gradualmente el intervalo de backoff. El intervalo real se selecciona aleatoriamente entre el 75 % y el 125 % del intervalo base actual. Tablestore SDK for Java 5.4.0 y versiones posteriores admiten este parámetro.
readMaxTimesPerRound (opcional)	int	El número máximo de llamadas a `ReadRecords` en una ronda de pipeline. El valor predeterminado es `1`.
readMaxBytesPerRound (opcional)	int	La cantidad máxima de datos extraídos en una ronda de pipeline, en bytes. El valor predeterminado es `4194304`, es decir, 4 MiB. La ronda se detiene al alcanzar este valor o `readMaxTimesPerRound`.
enableClosingChannelDetect (opcional)	boolean	Especifica si se detectan en tiempo real los canales en estado `CLOSING`. Un canal en estado `CLOSING` se está migrando de un cliente a otro. Tablestore SDK for Java 5.13.13 y versiones posteriores admiten este parámetro. El valor predeterminado es `true` en la versión 5.17.0 y posteriores. Si la detección está deshabilitada, la migración de canales puede bloquearse y el consumo puede interrumpirse cuando existen muchos canales pero los recursos del cliente son insuficientes.

Si inicia varias instancias de TunnelWorker en la misma máquina, puede reutilizar un único objeto TunnelWorkerConfig para compartir los pools de hilos de lectura y procesamiento. Una vez que todos los workers se detengan, llame a config.shutdown() solo una vez.

Datos de la función de callback

El método process recibe un objeto ProcessRecordsInput que contiene los siguientes campos.

Campo	Tipo	Descripción
records	`List<StreamRecord>`	Los registros extraídos en el lote actual. Llame a `getRecords()` para obtenerlos.
nextToken	String	El token para el siguiente lote. Llame a `getNextToken()` para obtenerlo. `TunnelWorker` utiliza automáticamente este valor para continuar extrayendo datos y registrando checkpoints.
traceId	String	El ID de rastreo de la solicitud de extracción actual. Llame a `getTraceId()` para obtenerlo.
channelId	String	El ID del canal al que pertenece el lote actual. Llame a `getChannelId()` para obtenerlo. Llame a `getPartitionId()` para obtener el ID de partición a partir del ID del canal.

Ejemplos de escenarios

Ajuste de parámetros de consumo

Si el throughput de consumo o el uso de memoria no cumplen con sus requisitos, ajuste los intervalos de heartbeat y checkpoint, la concurrencia de canales, la cantidad y el tamaño de las extracciones por ronda, así como el intervalo de backoff para las extracciones incrementales.

TunnelWorkerConfig config =
        new TunnelWorkerConfig(new SimpleProcessor());
config.setHeartbeatIntervalInSec(10);
config.setHeartbeatTimeoutInSec(60);
config.setCheckpointIntervalInMillis(10_000);
config.setMaxChannelParallel(16);
config.setReadMaxTimesPerRound(4);
config.setReadMaxBytesPerRound(8 * 1024 * 1024);
config.setMaxRetryIntervalInMillis(3_000);