kafka消费者异步手动提交偏移量方法怎么写

Kafka消费者异步手动提交偏移量一般需要以下几个步骤：

1. 创建Kafka消费者并订阅topic。

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test-group");
props.put("enable.auto.commit", "false");
props.put("auto.offset.reset", "earliest");
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);

consumer.subscribe(Collections.singletonList("test-topic"));

2. 在消费消息时手动提交偏移量。

try {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
        for (ConsumerRecord<String, String> record : records) {
            // 消费消息
            System.out.printf("topic = %s, partition = %s, offset = %d, key = %s, value = %s\n",
                    record.topic(), record.partition(), record.offset(), record.key(), record.value());
        }
        // 异步提交偏移量
        consumer.commitAsync(new OffsetCommitCallback() {
            @Override
            public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
                if (exception != null) {
                    System.err.println("Commit failed for " + offsets + ", " + exception.getMessage());
                }
            }
        });
    }
} finally {
    // 关闭消费者
    consumer.close();
}

在上面的代码中，我们使用`consumer.commitAsync()`方法异步提交偏移量，并在回调函数中处理提交结果和异常情况。需要注意的是，如果在提交偏移量时出现异常，我们可以选择重试或者将消息进行处理，具体取决于业务需求。

另外，我们也可以使用`consumer.commitSync()`方法同步提交偏移量，但是这种方式可能会影响消费性能，因为它会阻塞消费线程。因此，异步提交偏移量是更常用的方式。