提供了一个有n个分区的主题。spring的kafkaconsumer监听器有没有办法一次从同一个/一个分区监听多条消息?
我试过了 ConcurrentKafkaListenerContainerFactory
通过设置 setBatchListener(true);
但是消费者已经开始消费来自不同分区的多条消息,而不是一条。
public class BatchReceiverConfig {
@Value("${kafka.bootstrap-servers}")
private String bootstrapServers;
@Bean
public Map<String, Object> consumerConfigs() {
Map<String, Object> props = new HashMap<>();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.GROUP_ID_CONFIG, "batch4");
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
// maximum records per batch receive
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "10");
return props;
}
@Bean
public ConsumerFactory<String, String> consumerFactory() {
return new DefaultKafkaConsumerFactory<>(consumerConfigs());
}
@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<String, String> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory());
factory.setConcurrency(2);
// enable batch listeners
// factory.setBatchListener(true);
return factory;
}
@Bean
public BatchReceiver receiver() {
return new BatchReceiver();
}
}
/* Listener */
@KafkaListener(id = "batch-listener", topics = TOPIC_TEST_BATCH)
public void receive(List<String> data,
@Header(KafkaHeaders.RECEIVED_PARTITION_ID) List<Integer> partitions,
@Header(KafkaHeaders.OFFSET) List<Long> offsets) {
LOGGER.info("start of batch receive");
for (int i = 0; i < data.size(); i++) {
LOGGER.info("received message='{}' with partition-offset='{}'", data.get(i),
partitions.get(i) + "-" + offsets.get(i));
// handle message
latch.countDown();
}
LOGGER.info("end of batch receive");
}
2条答案
按热度按时间uinbv5nw1#
如果您使用组管理,kafka将为您的使用者分配分区。每个使用者(或使用并发时的线程)将获得0、1或更多分区,具体取决于使用者和分区的数量。
如果您希望某个特定的使用者只从一个分区获取消息,那么您必须自己分配分区,而不是让kafka进行分配。
如果您使用并发容器;您必须分配与并发相同数量的分区,因此每个线程只能获得一个分区。
0tdrvxhp2#
如果一个组中只有一个使用者,则可以设置
concurrencyLevel
那是什么ConcurrentKafkaListener
至n
.