As traduções são geradas por tradução automática. Em caso de conflito entre o conteúdo da tradução e da versão original em inglês, a versão em inglês prevalecerá.

Desenvolva consumidores com KCL Java

Pré-requisitos

Antes de começar a usar o KCL 3.x, verifique se você tem o seguinte:

KCLcoleta métricas de CPU utilização, como a CPU utilização do host de computação em que os trabalhadores estão executando, para equilibrar a carga e alcançar um nível uniforme de utilização de recursos entre os trabalhadores. Para permitir KCL a coleta de métricas de CPU utilização dos trabalhadores, você deve atender aos seguintes pré-requisitos:

Amazon Elastic Compute Cloud(AmazonEC2)

Amazon Elastic Container Service (AmazonECS) na Amazon EC2

Amazon ECS em AWS Fargate

Amazon Elastic Kubernetes Service (Amazon) na Amazon EKS EC2

Amazon EKS em AWS Fargate

Instalar e adicionar dependências

Se você estiver usando o Maven, adicione a seguinte dependência ao seu pom.xml arquivo. Certifique-se de ter substituído 3.x.x pela versão mais recenteKCL.

<dependency>
    <groupId>software.amazon.kinesis</groupId>
    <artifactId>amazon-kinesis-client</artifactId>
    <version>3.x.x</version> <!-- Use the latest version -->
</dependency>

Se você estiver usando o Gradle, adicione o seguinte ao seu build.gradle arquivo. Certifique-se de ter substituído 3.x.x pela versão mais recenteKCL.

implementation 'software.amazon.kinesis:amazon-kinesis-client:3.x.x'

Você pode verificar a versão mais recente do KCL no Repositório Central do Maven.

Implementar o consumidor

Um aplicativo para KCL consumidores consiste nos seguintes componentes principais:

RecordProcessor

RecordProcessor é o componente principal em que reside sua lógica de negócios para processar registros de stream de dados do Kinesis. Ele define como seu aplicativo processa os dados que recebe do stream do Kinesis.

Principais responsabilidades:

Veja a seguir um exemplo de implementação:

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
import software.amazon.kinesis.exceptions.InvalidStateException;
import software.amazon.kinesis.exceptions.ShutdownException;
import software.amazon.kinesis.lifecycle.events.*;
import software.amazon.kinesis.processor.ShardRecordProcessor;

public class SampleRecordProcessor implements ShardRecordProcessor {
    private static final String SHARD_ID_MDC_KEY = "ShardId";
    private static final Logger log = LoggerFactory.getLogger(SampleRecordProcessor.class);
    private String shardId;

    @Override
    public void initialize(InitializationInput initializationInput) {
        shardId = initializationInput.shardId();
        MDC.put(SHARD_ID_MDC_KEY, shardId);
        try {
            log.info("Initializing @ Sequence: {}", initializationInput.extendedSequenceNumber());
        } finally {
            MDC.remove(SHARD_ID_MDC_KEY);
        }
    }

    @Override
    public void processRecords(ProcessRecordsInput processRecordsInput) {
        MDC.put(SHARD_ID_MDC_KEY, shardId);
        try {
            log.info("Processing {} record(s)", processRecordsInput.records().size());
            processRecordsInput.records().forEach(r -> 
                log.info("Processing record pk: {} -- Seq: {}", r.partitionKey(), r.sequenceNumber())
            );
            
            // Checkpoint periodically
            processRecordsInput.checkpointer().checkpoint();
        } catch (Throwable t) {
            log.error("Caught throwable while processing records. Aborting.", t);
        } finally {
            MDC.remove(SHARD_ID_MDC_KEY);
        }
    }

    @Override
    public void leaseLost(LeaseLostInput leaseLostInput) {
        MDC.put(SHARD_ID_MDC_KEY, shardId);
        try {
            log.info("Lost lease, so terminating.");
        } finally {
            MDC.remove(SHARD_ID_MDC_KEY);
        }
    }

    @Override
    public void shardEnded(ShardEndedInput shardEndedInput) {
        MDC.put(SHARD_ID_MDC_KEY, shardId);
        try {
            log.info("Reached shard end checkpointing.");
            shardEndedInput.checkpointer().checkpoint();
        } catch (ShutdownException | InvalidStateException e) {
            log.error("Exception while checkpointing at shard end. Giving up.", e);
        } finally {
            MDC.remove(SHARD_ID_MDC_KEY);
        }
    }

    @Override
    public void shutdownRequested(ShutdownRequestedInput shutdownRequestedInput) {
        MDC.put(SHARD_ID_MDC_KEY, shardId);
        try {
            log.info("Scheduler is shutting down, checkpointing.");
            shutdownRequestedInput.checkpointer().checkpoint();
        } catch (ShutdownException | InvalidStateException e) {
            log.error("Exception while checkpointing at requested shutdown. Giving up.", e);
        } finally {
            MDC.remove(SHARD_ID_MDC_KEY);
        }
    }
}

Veja a seguir uma explicação detalhada de cada método usado no exemplo:

inicializar () InitializationInput initializationInput

processRecords(ProcessRecordsInput processRecordsInput)

leaseLost(LeaseLostInput leaseLostInput)

shardEnded(ShardEndedInput shardEndedInput)

shutdownRequested(ShutdownRequestedInput shutdownRequestedInput)

RecordProcessorFactory

RecordProcessorFactory é responsável pela criação de novas RecordProcessor instâncias. KCLusa essa fábrica para criar um novo RecordProcessor para cada fragmento que o aplicativo precisa processar.

Principais responsabilidades:

Veja a seguir um exemplo de implementação:

import software.amazon.kinesis.processor.ShardRecordProcessor;
import software.amazon.kinesis.processor.ShardRecordProcessorFactory;

public class SampleRecordProcessorFactory implements ShardRecordProcessorFactory {
    @Override
    public ShardRecordProcessor shardRecordProcessor() {
        return new SampleRecordProcessor();
    }
}

Neste exemplo, a fábrica cria um novo SampleRecordProcessor cada vez que shardRecordProcessor () é chamado. Você pode estender isso para incluir qualquer lógica de inicialização necessária.

Scheduler

O Scheduler é um componente de alto nível que coordena todas as atividades do KCL aplicativo. É responsável pela orquestração geral do processamento de dados.

Principais responsabilidades:

Normalmente, o agendador é criado e iniciado no aplicativo principal. Você pode verificar o exemplo de implementação do Scheduler na seção a seguir, Main Consumer Application.

Aplicação principal do consumidor

O aplicativo principal do consumidor une todos os componentes. É responsável por configurar o KCL consumidor, criar os clientes necessários, configurar o Agendador e gerenciar o ciclo de vida do aplicativo.

Principais responsabilidades:

Veja a seguir um exemplo de implementação:

import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.cloudwatch.CloudWatchAsyncClient;
import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient;
import software.amazon.awssdk.services.kinesis.KinesisAsyncClient;
import software.amazon.kinesis.common.ConfigsBuilder;
import software.amazon.kinesis.common.KinesisClientUtil;
import software.amazon.kinesis.coordinator.Scheduler;
import java.util.UUID;

public class SampleConsumer {
    private final String streamName;
    private final Region region;
    private final KinesisAsyncClient kinesisClient;

    public SampleConsumer(String streamName, Region region) {
        this.streamName = streamName;
        this.region = region;
        this.kinesisClient = KinesisClientUtil.createKinesisAsyncClient(KinesisAsyncClient.builder().region(this.region));
    }

    public void run() {
        DynamoDbAsyncClient dynamoDbAsyncClient = DynamoDbAsyncClient.builder().region(region).build();
        CloudWatchAsyncClient cloudWatchClient = CloudWatchAsyncClient.builder().region(region).build();
        
        ConfigsBuilder configsBuilder = new ConfigsBuilder(
            streamName, 
            streamName, 
            kinesisClient, 
            dynamoDbAsyncClient,
            cloudWatchClient, 
            UUID.randomUUID().toString(), 
            new SampleRecordProcessorFactory()
        );

        Scheduler scheduler = new Scheduler(
            configsBuilder.checkpointConfig(),
            configsBuilder.coordinatorConfig(),
            configsBuilder.leaseManagementConfig(),
            configsBuilder.lifecycleConfig(),
            configsBuilder.metricsConfig(),
            configsBuilder.processorConfig(),
            configsBuilder.retrievalConfig()
        );

        Thread schedulerThread = new Thread(scheduler);
        schedulerThread.setDaemon(true);
        schedulerThread.start();
    }

    public static void main(String[] args) {
        String streamName = "your-stream-name"; // replace with your stream name
        Region region = Region.US_EAST_1; // replace with your region
        new SampleConsumer(streamName, region).run();
    }
}

KCLcria um consumidor Enhanced Fan-out (EFO) com taxa de transferência dedicada por padrão. Para obter mais informações sobre o Enhanced Fan-out, consulte. Desenvolva consumidores de distribuição aprimorados com taxa de transferência dedicada Se você tiver menos de 2 consumidores ou não precisar de atrasos de propagação de leitura abaixo de 200 ms, defina a seguinte configuração no objeto do agendador para usar consumidores de taxa de transferência compartilhada:

configsBuilder.retrievalConfig().retrievalSpecificConfig(new PollingConfig(streamName, kinesisClient))

O código a seguir é um exemplo de criação de um objeto agendador que usa consumidores de taxa de transferência compartilhada:

Importações:

import software.amazon.kinesis.retrieval.polling.PollingConfig;

Código:

Scheduler scheduler = new Scheduler(
            configsBuilder.checkpointConfig(),
            configsBuilder.coordinatorConfig(),
            configsBuilder.leaseManagementConfig(),
            configsBuilder.lifecycleConfig(),
            configsBuilder.metricsConfig(),
            configsBuilder.processorConfig(),
            configsBuilder.retrievalConfig().retrievalSpecificConfig(new PollingConfig(streamName, kinesisClient))
        );/