kafka获取消息并手动提交

### 获取消息并手动提交 Offset 在 Kafka 中，通过 `KafkaConsumer` 类可以实现消费消息，并且可以通过调用手动提交偏移量的方法来控制何时确认已处理的消息。对于手动提交，存在两种主要方式：同步提交 (`commitSync`) 和异步提交 (`commitAsync`)。 #### 同步提交 (Commit Sync) 当使用 `commitSync()` 方法时，在每次成功发送请求之后都会等待服务器响应。这意味着如果发生错误，则会抛出异常；如果没有收到任何错误反馈，则认为此次提交已经完成[^3]。 ```java consumer.commitSync(); ``` 此方法适用于希望确保每个批次的数据都被可靠地记录下来的应用场景下。它提供了更强的一致性和可靠性保障，因为程序会在继续之前验证每一步骤的结果。 #### 异步提交 (Commit Async) 另一方面，`commitAsync()` 不会阻塞当前线程直到操作结束，而是立即返回给应用程序执行其他任务。这有助于提高吞吐量，但也可能带来一些挑战——比如在网络分区期间可能会丢失未被持久化的位移数据。 ```java consumer.commitAsync(new OffsetCommitCallback() { @Override public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) { if(exception != null){ System.out.println("Failed to commit offsets due to : "+exception.getMessage()); }else{ System.out.println("Offsets committed successfully"); } } }); ``` 为了防止潜在的问题，建议结合两者优点的做法是在正常情况下采用异步模式以获得更好的性能表现，而在定期间隔内加入一次性的同步提交作为额外的安全措施[^2]。 #### 完整示例代码 下面是一个完整的 Java 例子展示了如何配置消费者属性、订阅主题以及循环读取消息的同时进行手动提交： ```java Properties props = new Properties(); props.put("bootstrap.servers", "localhost:9092"); props.put("group.id", "test-group"); // 设置为false表示禁用自动提交 props.put("enable.auto.commit", "false"); props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); final KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props); consumer.subscribe(Arrays.asList("topic-name")); try { while (true) { ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100)); for (ConsumerRecord<String, String> record : records) { // Process the message here... System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value()); } // Manually commit after processing all messages from this poll. consumer.commitSync(); // or use commitAsync(callback) } } finally { consumer.close(); } ``` 上述代码片段中设置了 `enable.auto.commit=false` 来关闭默认的自动提交行为，从而允许开发者完全掌控偏移量管理逻辑[^1]。