kafka手动提交offset

　　Kafka 的消费者可以使用两种方式来提交消费位移(offset)：自动提交和手动提交。自动提交是由 Kafka 客户端自动定期提交位移，而手动提交则需要应用程序显式地调用 API 来提交位移。手动提交位移的方式可以更精细地控制消费位移，以及避免因自动提交位移而产生的数据丢失或重复消费等问题。

　　下面是使用 Kafka Java API 手动提交位移的一些示例代码：

　　1.启用手动提交位移：

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test-group");
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");
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

 　　在创建 KafkaConsumer 对象时，将 enable.auto.commit 属性设置为 false，以关闭自动提交位移的功能。

　　2.手动提交位移：

while (true) {
  ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
  for (ConsumerRecord<String, String> record : records) {
    // 处理消息
  }
  consumer.commitSync(); // 手动提交位移
}

 　　在消费消息后，调用 commitSync() 方法来手动提交位移。如果需要批量提交位移，可以使用 commitSync(Map<topicpartition, offsetandmetadata=""> offsets) 方法来提交指定的分区和位移信息。

　　需要注意的是，手动提交位移需要在适当的时机进行提交，以确保数据不会丢失或重复消费。一般来说，可以在消费一批消息后，或者在处理完一段业务逻辑后，再进行位移提交。同时，还需要注意位移的提交顺序，以保证数据的一致性。