spring-kafka手动提交

Spring Kafka中的手动提交是指在消费者应用程序中显式地将消费的位置信息（offset）发送回Kafka服务器，以更新消费者的偏移量记录。这通常发生在以下几种情况：

1. **批量处理完成**：如果你的应用需要对接收到的消息进行批处理，处理完成后，你需要确认这一批次已经成功处理，这时可以手动提交对应的offset。

2. **异常恢复**：如果因为某个错误导致消息无法正常处理，消费者可以选择重新开始从最近一次提交的位置读取，而不是从头开始，这就需要用到手动提交。

3. **幂等性和一致性保证**：某些场景下，消费者需要确保处理消息的结果是幂等的，手动提交能帮助跟踪每个消息的处理状态。

在Spring Kafka中，手动提交通常是通过`org.springframework.kafka.annotation.KafkaListener`的`@CommitPolicy`注解设置为`COMMIT_ON_SUCCESS`或使用`KafkaTemplate`的`sendAndForget()`方法配合`commitOffset()`方法来实现的。例如：

@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> containerFactory() {
    ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    // 设置手动提交策略
    factory.setCommitOnCompletion(true);
    return factory;
}

// 或者在监听器方法中手动提交
@Autowired
private MessageListenerContainer messageListenerContainer;

@KafkaListener(topics = "my-topic")
public void listen(@Payload String message) {
    try {
        // ...业务处理...
        messageListenerContainer.commitMessage(message, true); // 手动提交
    } catch (Exception e) {
        // 异常处理...
    }
}