Spring-Kafka 1.3.x批量监听器安全消费方案问询

针对你使用Spring-Kafka 1.3.3/1.3.4版本的需求，我整理了一套基于Spring容器原生能力的安全批量消费方案，确保异常（比如网络中断）时不会跳过消息或消息组，不用自定义框架就能实现：

核心思路 #

要实现安全批量消费，关键在于控制offset提交时机——只有当整个批次的消息处理成功后才提交offset；如果发生异常，不提交offset，让Kafka容器在下一次拉取时重新获取这批消息。同时结合重试机制处理瞬时异常，避免不必要的重复拉取。

具体配置与实现步骤 #

1. 配置消费者工厂，关闭自动提交offset #

首先要关闭Kafka的自动offset提交，改为由Spring容器手动控制：

@Bean
public ConsumerFactory<String, String> consumerFactory() {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "你的Kafka地址");
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "你的消费组ID");
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    
    // 核心：关闭自动提交offset
    props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
    // 设置批量拉取的消息数量（根据你的业务调整）
    props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 100);
    // 调整最大拉取间隔，避免长时间处理导致rebalance（比如设置5分钟）
    props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 300000);
    
    return new DefaultKafkaConsumerFactory<>(props);
}

2. 配置ConcurrentMessageListenerContainer，使用批量监听器+手动确认 #

使用AcknowledgingBatchMessageListener接收批量消息，并通过Acknowledgment对象控制offset提交：

@Bean
public ConcurrentMessageListenerContainer<String, String> kafkaListenerContainer() {
    ContainerProperties containerProps = new ContainerProperties("你的目标Topic");
    
    // 使用批量消息监听器，同时接收确认对象
    containerProps.setMessageListener(new AcknowledgingBatchMessageListener<String, String>() {
        @Override
        public void onMessage(List<ConsumerRecord<String, String>> records, Acknowledgment acknowledgment) {
            int retryAttempts = 3; // 设置重试次数，处理瞬时异常
            boolean processedSuccessfully = false;
            
            while (retryAttempts > 0 && !processedSuccessfully) {
                try {
                    // 执行你的批量消息处理逻辑
                    processBatchOfMessages(records);
                    
                    // 处理成功后，批量提交offset
                    acknowledgment.acknowledge();
                    processedSuccessfully = true;
                    log.info("Batch processed successfully, offset committed");
                } catch (Exception e) {
                    retryAttempts--;
                    log.warn("Batch processing failed, remaining retries: {}", retryAttempts, e);
                    
                    if (retryAttempts == 0) {
                        // 重试耗尽后，将消息转存到死信队列（避免阻塞正常消费）
                        // 这里需要你自己实现死信队列的发送逻辑
                        sendToDeadLetterQueue(records);
                        // 提交offset，避免重复处理这批失败的消息
                        acknowledgment.acknowledge();
                        log.error("All retries exhausted, batch sent to dead letter queue");
                    }
                    
                    // 重试间隔，避免频繁重试
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException ie) {
                        Thread.currentThread().interrupt();
                    }
                }
            }
        }
    });
    
    // 设置手动确认模式
    containerProps.setAckMode(ContainerProperties.AckMode.MANUAL);
    // 设置并发消费线程数（根据你的服务器资源调整）
    ConcurrentMessageListenerContainer<String, String> container = 
        new ConcurrentMessageListenerContainer<>(consumerFactory(), containerProps);
    container.setConcurrency(3);
    
    return container;
}

3. 关键细节说明 #

• offset提交逻辑：只有当整个批次处理成功（或重试耗尽后转存死信）才提交offset，确保不会丢失或跳过消息。
• 重试机制：针对网络中断这类瞬时异常，通过重试提高成功率，减少重复拉取的次数。
• 死信队列处理：对于永久异常（比如消息格式错误），转存死信队列后提交offset，避免阻塞消费组的正常消费。
• 消费组保障：只要offset提交正确，同一消费组的消费者会从正确的offset位置开始消费，不会跳过整个消息组。

注意事项 #

• 如果你不需要重试逻辑，可以直接在异常时不提交offset，容器会自动在下一次拉取时重新获取这批消息，但要注意避免因永久异常导致无限循环。
• 1.3.x版本没有原生的死信容器支持，所以死信队列需要你自己实现（比如用KafkaTemplate发送到指定的死信Topic）。

内容的提问来源于stack exchange，提问作者Mich Betancourt