使用Kafka消费者池是否正确?

问题描述 投票:0回答:1

有时,我需要从特定分区读取具有相同主题的特定偏移的记录。

我每次都可以创建新的kafka消费者。但是,我可以创建消费者池并以这种方式使用它:

List<KafkaConsumer> consumers = new ArrayList<>();

// acquire consumer
KafkaConsumer consumer = consumers.get(0);
TopicPartition topicPartition = new TopicPartition("my-topic", 42);
consumer.assign(List.of(topicPartition));
consumer.seek(topicPartition, 13);

ConsumerRecords records = consumer.poll(0);
// process records
// .....

// release consumer
consumer.unsubscribe();

我应该建立消费者群吗?或者它没有效果,我应该为每个使用创造新的消费者。

java apache-kafka kafka-consumer-api pool
1个回答
2
投票

你只需要一个消费者。只需取消订阅并将其重新分配给另一个TopicPartition

String topic = "my-topic";
int partition = 42;
int offset = 13;
boolean running = true;

while(running) {
    TopicPartition topicPartition = new TopicPartition(topic, partition);
    consumer.assign(List.of(topicPartition));
    consumer.seek(topicPartition, offset);

    ConsumerRecords records = consumer.poll(0);
    // process records
    // .....

    // release consumer
    consumer.unsubscribe();
    // Change topic, partition, offset as needed
}
© www.soinside.com 2019 - 2024. All rights reserved.