Kafka消费者手动提交实现过程

Kafka消费者手动提交偏移量的实现步骤如下：

1. 设置消费者配置参数

首先，需要设置以下消费者配置参数：

- enable.auto.commit=false：禁用自动提交偏移量
- max.poll.records：每次拉取的最大记录数
- auto.offset.reset：当消费者组中没有存储偏移量或者偏移量不存在时，从哪里开始消费消息。可以设置为 earliest 或者 latest。

示例代码：

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test-group");
props.put("enable.auto.commit", "false");
props.put("max.poll.records", "500");
props.put("auto.offset.reset", "earliest");

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

2. 手动提交偏移量

在消费消息的过程中，需要手动提交偏移量。可以在每次拉取到消息之后，处理完消息后立即提交偏移量，也可以在一段时间内累积一定量的消息后再提交偏移量。

示例代码：

try {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, String> record : records) {
            // 处理消息
        }
        consumer.commitSync(); // 同步提交偏移量
    }
} finally {
    consumer.close();
}

在上面的代码中，每次拉取到消息后，处理完消息后立即提交偏移量，使用的是同步提交偏移量的方式。如果使用异步提交偏移量的方式，可以使用 `commitAsync()` 方法。

需要注意的是，手动提交偏移量的方式需要确保偏移量提交的正确性和可靠性，否则可能会导致消息被重复消费或者消息丢失的问题。