Terjemahan disediakan oleh mesin penerjemah. Jika konten terjemahan yang diberikan bertentangan dengan versi bahasa Inggris aslinya, utamakan versi bahasa Inggris.

Kembangkan konsumen dengan KCL di Jawa

Prasyarat

Sebelum Anda mulai menggunakan KCL 3.x, pastikan Anda memiliki yang berikut:

KCL mengumpulkan metrik pemanfaatan CPU seperti pemanfaatan CPU dari host komputasi yang dijalankan pekerja untuk menyeimbangkan beban guna mencapai tingkat pemanfaatan sumber daya yang merata di seluruh pekerja. Untuk mengaktifkan KCL mengumpulkan metrik pemanfaatan CPU dari pekerja, Anda harus memenuhi prasyarat berikut:

Amazon Elastic Compute Cloud(Amazon EC2)

Layanan Kontainer Elastis Amazon (Amazon ECS) di Amazon EC2

Amazon ECS aktif AWS Fargate

Layanan Kubernetes Elastis Amazon (Amazon EKS) di Amazon EC2

Amazon EKS pada AWS Fargate

Instal dan tambahkan dependensi

Jika Anda menggunakan Maven, tambahkan dependensi berikut ke pom.xml file Anda. Pastikan Anda mengganti 3.x.x ke versi KCL terbaru.

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

Jika Anda menggunakan Gradle, tambahkan yang berikut ini ke build.gradle file Anda. Pastikan Anda mengganti 3.x.x ke versi KCL terbaru.

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

Anda dapat memeriksa versi terbaru KCL di Maven Central Repository.

Menerapkan konsumen

Aplikasi konsumen KCL terdiri dari komponen-komponen kunci berikut:

RecordProcessor

RecordProcessor adalah komponen inti tempat logika bisnis Anda untuk memproses catatan aliran data Kinesis berada. Ini mendefinisikan bagaimana aplikasi Anda memproses data yang diterimanya dari aliran Kinesis.

Tanggung jawab utama:

Berikut ini menunjukkan contoh implementasi:

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

Berikut ini adalah penjelasan rinci dari setiap metode yang digunakan dalam contoh:

menginisialisasi (InitializationInputInitializationInput)

ProsesRecords (ProcessRecordsInputprosesRecordsInput)

LeaseLost (sewa) LeaseLostInput LostInput

ShardenDed (pecahan) ShardEndedInput EndedInput

ShutdownRequested (shutdown) ShutdownRequestedInput RequestedInput

RecordProcessorFactory

RecordProcessorFactory bertanggung jawab untuk membuat RecordProcessor instance baru. KCL menggunakan pabrik ini untuk membuat yang baru RecordProcessor untuk setiap pecahan yang perlu diproses aplikasi.

Tanggung jawab utama:

Berikut ini adalah contoh implementasi:

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

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

Dalam contoh ini, pabrik membuat yang baru SampleRecordProcessor setiap kali shard RecordProcessor () dipanggil. Anda dapat memperluas ini untuk memasukkan logika inisialisasi yang diperlukan.

Penjadwal

Scheduler adalah komponen tingkat tinggi yang mengoordinasikan semua aktivitas aplikasi KCL. Ini bertanggung jawab atas keseluruhan orkestrasi pemrosesan data.

Tanggung jawab utama:

Scheduler biasanya dibuat dan dimulai di Aplikasi Utama. Anda dapat memeriksa contoh implementasi Scheduler di bagian berikut, Aplikasi Konsumen Utama.

Aplikasi Konsumen Utama

Aplikasi Konsumen Utama mengikat semua komponen bersama-sama. Ini bertanggung jawab untuk mengatur konsumen KCL, membuat klien yang diperlukan, mengonfigurasi Penjadwal, dan mengelola siklus hidup aplikasi.

Tanggung jawab utama:

Berikut ini adalah contoh implementasi:

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 menciptakan konsumen Enhanced Fan-out (EFO) dengan throughput khusus secara default. Untuk informasi selengkapnya tentang Enhanced Fan-out, lihatKembangkan konsumen fan-out yang ditingkatkan dengan throughput khusus. Jika Anda memiliki kurang dari 2 konsumen atau tidak memerlukan penundaan propagasi baca di bawah 200 ms, Anda harus mengatur konfigurasi berikut di objek penjadwal untuk menggunakan konsumen throughput bersama:

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

Kode berikut adalah contoh pembuatan objek penjadwal yang menggunakan konsumen throughput bersama:

Impor:

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

Kode:

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