Le traduzioni sono generate tramite traduzione automatica. In caso di conflitto tra il contenuto di una traduzione e la versione originale in Inglese, quest'ultima prevarrà.

Sviluppa i consumatori con KCL in Java

Prerequisiti

Prima di iniziare a usare KCL 3.x, assicurati di avere quanto segue:

KCL raccoglie le metriche di utilizzo della CPU, ad esempio l'utilizzo della CPU, dall'host di elaborazione su cui sono in esecuzione i lavoratori per bilanciare il carico e raggiungere un livello di utilizzo delle risorse uniforme tra i lavoratori. Per consentire a KCL di raccogliere le metriche sull'utilizzo della CPU dai lavoratori, devi soddisfare i seguenti prerequisiti:

Amazon Elastic Compute Cloud(Amazon EC2)

Amazon Elastic Container Service (Amazon ECS) su Amazon EC2

Amazon ECS attivo AWS Fargate

Amazon Elastic Kubernetes Service (Amazon EKS) su Amazon EC2

Amazon EKS su AWS Fargate

Installa e aggiungi dipendenze

Se stai usando Maven, aggiungi la seguente dipendenza al tuo file. pom.xml Assicurati di aver sostituito 3.x.x con l'ultima versione di KCL.

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

Se stai usando Gradle, aggiungi quanto segue al tuo file. build.gradle Assicurati di aver sostituito 3.x.x con l'ultima versione di KCL.

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

Puoi controllare l'ultima versione di KCL nel Maven Central Repository. https://search.maven.org/artifact/software.amazon.kinesis/amazon-kinesis-client

Implementa il consumatore

Un'applicazione consumer KCL è costituita dai seguenti componenti chiave:

RecordProcessor

RecordProcessor è il componente principale in cui risiede la logica aziendale per l'elaborazione dei record del flusso di dati Kinesis. Definisce il modo in cui l'applicazione elabora i dati che riceve dallo stream Kinesis.

Responsabilità principali:

Quanto segue mostra un esempio di implementazione:

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

Di seguito è riportata una spiegazione dettagliata di ciascun metodo utilizzato nell'esempio:

inizializza (InitializationInputinitializationInput)

ProcessRecords (ProcessRecordsInputprocessoRecordsInput)

LeaseLost (leasing) LeaseLostInput LostInput

shardEnded (shard) ShardEndedInput EndedInput

ShutdownRequestedInputRequestedInputShutdownRequested (spegnimento)

RecordProcessorFactory

RecordProcessorFactory è responsabile della creazione di nuove RecordProcessor istanze. KCL utilizza questa fabbrica per crearne una nuova RecordProcessor per ogni frammento che l'applicazione deve elaborare.

Responsabilità principali:

Di seguito è riportato un esempio di implementazione:

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

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

In questo esempio, il factory ne crea uno nuovo SampleRecordProcessor ogni volta che viene chiamato shard RecordProcessor (). È possibile estenderlo per includere qualsiasi logica di inizializzazione necessaria.

Scheduler

Scheduler è un componente di alto livello che coordina tutte le attività dell'applicazione KCL. È responsabile dell'orchestrazione complessiva dell'elaborazione dei dati.

Responsabilità chiave:

Lo Scheduler viene in genere creato e avviato nell'applicazione principale. È possibile controllare l'esempio di implementazione di Scheduler nella sezione seguente, Main Consumer Application.

Applicazione consumer principale

La Main Consumer Application unisce tutti i componenti. È responsabile della configurazione del consumatore KCL, della creazione dei client necessari, della configurazione dello Scheduler e della gestione del ciclo di vita dell'applicazione.

Responsabilità principali:

Di seguito è riportato un esempio di implementazione:

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

Per impostazione predefinita, KCL crea un consumatore Enhanced Fan-out (EFO) con throughput dedicato. Per ulteriori informazioni su Enhanced Fan-out, vedere. Sviluppa consumatori con fan-out migliorati con un throughput dedicato Se hai meno di 2 consumatori o non hai bisogno di ritardi di propagazione della lettura inferiori a 200 ms, devi impostare la seguente configurazione nell'oggetto scheduler per utilizzare i consumatori con throughput condiviso:

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

Il codice seguente è un esempio di creazione di un oggetto scheduler che utilizza consumer con throughput condiviso:

Importazioni:

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

Codice:

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