kafka java 代码

如何在消费者中手动提交偏移量？

在 Kafka 中，消费者通过偏移量（offset）来记录消费进度，以便在下次重新启动时继续从上次的位置消费。在消费者中，我们可以通过手动提交偏移量来控制消费进度。

在 Java 代码中，我们可以使用 Consumer#commitSync方法来手动提交偏移量。具体实现如下：

1. 创建 KafkaConsumer 实例

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

2. 订阅主题

consumer.subscribe(Collections.singletonList(topic));

3. 开始消费消息并手动提交偏移量

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

在代码中，我们首先创建一个 KafkaConsumer 实例，并订阅指定的主题。然后循环调用 consumer.poll 方法来拉取消息并处理。最后，在消费完成后调用 consumer.commitSync 方法手动提交偏移量。

需要注意的是，手动提交偏移量需要在每次消息消费完成后进行，否则可能导致重复消费或丢失消息的情况。另外，如果提交失败，需要根据具体情况进行处理，比如记录错误日志、重试等。