Kafka提交偏移量

使用Kafka Consumer API提交偏移量

Kafka Consumer API提供了两种方法来提交偏移量：自动提交和手动提交。自动提交通常是每隔一段时间或每个消息处理后自动提交偏移量，但这可能会导致数据重复或丢失的问题。手动提交则允许我们控制何时提交偏移量，从而避免上述问题。

以下是手动提交偏移量的代码示例：

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test-group");
props.put("enable.auto.commit", "false"); // 禁用自动提交

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("test-topic"));

try {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, String> record : records) {
            System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
        }
        consumer.commitSync(); // 手动提交偏移量
    }
} finally {
    consumer.close();
}