如何使用Quarkus在Kafka中设置同一主题中的多个消费者
我正在使用Quarkus框架构建一个Kafka消费者,它将读取带有3个分区的主题。下面的代码片段正在工作,但根据日志,我只是启动了具有3个分区的1个使用者。我现在的问题是,一旦我运行我的应用程序,我如何才能产生3个消费者。
@Incoming("topic-1")
public CompletionStage<Void> onMessage(KafkaRecord<String, String> message) throws IOException {
LOG.info("Kafka order message with value = {} arrived from topic {} ", message.getPayload(),
message.getTopic());
//JsonObject event = new JsonObject(message.getPayload());
try {
if (true) {
LOG.info("Kafka message: " + message);
}
} catch (Exception e) {
e.printStackTrace();
}
return message.ack();
}
请参阅示例日志:
信息[org.apa.kaf.cli.con.int.Consumer协调器](vert.x-kafka-Consumer-线程-0)[消费者客户端ID=测试消费者,组ID=kafka-检测-消费者]已完成第64代的组的分配:{testconsumer-bf6d314c-44e1-47b1-9439-fe4058951841=Assignment(partitions=[test_part-0,TEST_PART-1,TEST_PART-2])}
解决方案
如果您在Containers平台(Docker、K8s...)上运行应用程序然后,您可以横向扩展您的服务;否则,请使用不同的端口再次运行您的应用程序。
Kafka客户端启动时会被分配到某个分区,因此同一个客户端不能从多个Theme-Partition消费。
相关文章