我将我的Kafka侦听器的并发设置为1。
ConcurrentKafkaListenerContainerFactory<String, Map<String, Object>>
factory = new ConcurrentKafkaListenerContainerFactory<>();
factory.setConcurrency(conncurrency);
factory.setConsumerFactory(consumerFactory());
factory.setRetryTemplate(retryTemplate());
我正在听3个不同的主题
@KafkaListener(topics = "#{'${kafka.consumer.topic.name}'.split(',')}", containerFactory = "kafkaListenerContainerFactory")
public void listen(@Payload Map<String, Object> conciseMap,
@Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
@Header(KafkaHeaders.OFFSET) int offset,
Acknowledgment ack) {
processMessage(conciseMap,partition,offset,ack,false);
}
在这种情况下，侦听器是否会从第一个主题中读取一条消息，一旦处理完毕，便会从下一个主题中读取一条消息，依此类推？或者它将同时处理每个主题的1条消息。
如果它是前者，是否可以从所有主题中同时读取1条消息，而无需创建多个侦听器？
最佳答案
不能保证Kafka代理将如何在容器线程之间分配分区。如果您只有一个分区；它们可能将全部分配给同一容器线程。这就是我使用容器并发= 3运行测试时发生的事情...
2017-10-31 16：40：26.066 INFO 35202 --- [ntainer＃0-2-C-1] o.s.k.l.KafkaMessageListenerContainer：分配的分区：[]
2017-10-31 16：40：26.066 INFO 35202 --- [ntainer＃0-1-C-1] o.s.k.l.KafkaMessageListenerContainer：分配的分区：[]
2017-10-31 16：40：26.079 INFO 35202 --- [ntainer＃0-0-C-1] o.s.k.l.KafkaMessageListenerContainer：分配的分区：[bar-0，foo-0，baz-0]
每个主题有10个分区，我得到了这个分布...
2017-10-31 16：46：19.279信息35900 --- [ntainer＃0-1-C-1] osklKafkaMessageListenerContainer：分配的分区：[foo10-5，foo10-6，foo10-4，baz10-5，baz10 -4，baz10-6，bar10-5，bar10-4，bar10-6]
2017-10-31 16：46：19.279 INFO 35900 --- [ntainer＃0-0-C-1] osklKafkaMessageListenerContainer：分配的分区：[bar10-1，bar10-0，bar10-3，bar10-2，baz10 -1，baz10-0，baz10-3，baz10-2，foo10-3，foo10-1，foo10-2，foo10-0]
2017-10-31 16：46：19.279 INFO 35900 --- [ntainer＃0-2-C-1] osklKafkaMessageListenerContainer：分配的分区：[baz10-9，baz10-8，baz10-7，bar10-9，bar10 -8，foo10-9，bar10-7，foo10-7，foo10-8]
如您所见，每个主题的一些分区已分配给每个线程。但是其中两个线程共有9个分区，而一个线程有12个分区。
如果您想完全控制，我建议每个主题一个监听器。