我们是否可以使用spring kafka批处理监听器对消息进行一次处理？

Question

config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");    
config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
config.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 120000);
config.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 25);
config.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 4096000);
config.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, 120000);
config.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 600000);  
config.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 8192000);

但它不止一次地使用消息。有没有人面对过这个问题。此外，使用上述配置，使用者总是在一个批处理中只接收到一个消息。我尝试增加fetch.min.bytes和fetch.max.wait.ms，但没有任何影响。

在对ConcurrentKafkaListenerContainerFactory进行如下更改后，批处理配置的问题得到了解决：

ConcurrentKafkaListenerContainerFactory<String, String> factory =
    new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory());

factory.setBatchListener(true);
factory.getContainerProperties().setPollTimeout(3600000);

factory.getContainerProperties().SetackMode（org.springFramework.kafka.listner.ContainerProperties.ackMode.Manual）；factory.SetMessageConverter（新建BatchMessagingMessageConverter（stringJsonMessageConverter（）））；

Answer 1

要精确获取一次语义，必须使用事务。

然而，仅有一次语义只适用于

read from Kafka -> process -> write to Kafka

即使这样，它也只适用于整个流程（读/处理/写）。