Java에서 KCL을 사용하여 소비자 개발

사전 조건

KCL 3.x를 사용하여 시작하기 전에 다음이 있는지 확인합니다.

KCL은 워커가 작동 중인 컴퓨팅 호스트에서 CPU 사용률과 같은 CPU 사용률 지표를 수집하여 로드의 균형을 조정함으로써 워커 간에 리소스 사용률 수준을 균등하게 유지합니다. KCL이 워커로부터 CPU 사용률 지표를 수집할 수 있게 하려면 다음 사전 조건을 충족해야 합니다.

Amazon Elastic Compute Cloud(Amazon EC2)

Amazon EC2의 Amazon Elastic Container Service(Amazon ECS)

의 Amazon ECSAWS Fargate

Amazon EC2의 Amazon Elastic Kubernetes Service(Amazon EKS)

에 대한 Amazon EKSAWS Fargate

종속성 설치 및 추가

Maven을 사용하는 경우 pom.xml 파일에 다음 종속성을 추가합니다. 3.x.x를 최신 KCL 버전으로 교체했는지 확인합니다.

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

Gradle을 사용하는 경우 build.gradle 파일에 다음을 추가합니다. 3.x.x를 최신 KCL 버전으로 교체했는지 확인합니다.

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

Maven Central Repository에서 최신 버전의 KCL을 확인할 수 있습니다.

소비자 구현

KCL 소비자 애플리케이션은 다음과 같은 주요 구성 요소로 구성됩니다.

RecordProcessor

RecordProcessor는 Kinesis 데이터 스트림 레코드를 처리하는 비즈니스 로직이 상주하는 핵심 구성 요소입니다. 애플리케이션이 Kinesis 스트림에서 수신하는 데이터를 처리하는 방법을 정의합니다.

주요 책임:

다음은 구현 예제를 보여줍니다.

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

다음은 예제에서 사용된 각 메서드에 대한 상세 설명입니다.

initialize(InitializationInput initializationInput)

processRecords(ProcessRecordsInput processRecordsInput)

leaseLost(LeaseLostInput leaseLostInput)

shardEnded(ShardEndedInput shardEndedInput)

shutdownRequested(ShutdownRequestedInput shutdownRequestedInput)

RecordProcessorFactory

RecordProcessorFactory는 새 RecordProcessor 인스턴스를 생성하는 역할을 합니다. KCL은 이 팩토리를 사용하여 애플리케이션이 처리해야 하는 각 샤드에 대해 새 RecordProcessor를 생성합니다.

주요 책임:

다음은 구현 예제입니다.

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

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

이 예제에서는 shardRecordProcessor()가 호출될 때마다 팩토리에서 새 SampleRecordProcessor를 생성합니다. 필요한 초기화 로직을 포함하도록 이를 확장할 수 있습니다.

스케줄러

스케줄러는 KCL 애플리케이션의 모든 활동을 조정하는 상위 수준 구성 요소입니다. 데이터 처리의 전반적인 오케스트레이션을 담당합니다.

주요 책임:

스케줄러는 일반적으로 기본 애플리케이션에서 생성되고 시작됩니다. 스케줄러의 구현 예제는 기본 소비자 애플리케이션 섹션에서 확인할 수 있습니다.

기본 소비자 애플리케이션

기본 소비자 애플리케이션은 모든 구성 요소를 하나로 묶어줍니다. KCL 소비자 설정, 필요한 클라이언트 생성, 스케줄러 구성, 애플리케이션의 수명 주기 관리를 담당합니다.

주요 책임:

다음은 구현 예제입니다.

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

KCL은 기본적으로 전용 처리량으로 향상된 팬아웃(EFO) 소비자를 생성합니다. 향상된 팬아웃에 대한 자세한 내용은 전용 처리량으로 향상된 팬아웃 소비자 개발 섹션을 참조하세요. 소비자가 2명 미만이거나 200ms 미만의 읽기 전파 지연이 필요하지 않은 경우, 공유 처리량 소비자를 사용하도록 스케줄러 객체에서 다음 구성을 설정해야 합니다.

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

다음 코드는 공유 처리량 소비자를 사용하는 스케줄러 객체를 생성하는 예제입니다.

가져옵니다.

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

코드

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