117笔记问答在Kafka中，消费者进行消息确认的方式主要有两种：自动确认和手动确认。下面分别介绍这两种方式的具体实现。
- 自动确认：
在自动确认模式下，消费者在成功消费消息后，会自动向Kafka发送一个确认信号，表示该消息已经被成功处理。这种方式的优点是简化了消费者的代码逻辑，但缺点是一旦消费者崩溃或者出现其他问题，可能会导致一些消息被丢失。
要实现自动确认，需要在创建消费者时设置enable.auto.commit属性为true。例如，在Java客户端库中，可以通过以下方式创建一个自动确认的消费者：
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "my-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("enable.auto.commit", "true");
KafkaConsumer consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("my-topic"));
- 手动确认：
与自动确认相反，手动确认模式要求消费者在成功消费消息后，需要显式地向Kafka发送一个确认信号。这种方式的优点是可以更好地控制消息的处理流程，避免因消费者崩溃等问题导致的消息丢失。但缺点是需要编写额外的代码来处理消息确认。
要实现手动确认，需要在创建消费者时设置enable.auto.commit属性为false，并实现一个org.apache.kafka.clients.consumer.ConsumerAcknowledgment接口的类来处理消息确认。例如，在Java客户端库中，可以通过以下方式创建一个手动确认的消费者：
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "my-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("enable.auto.commit", "false");
KafkaConsumer consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("my-topic"));
// 创建一个手动确认的处理器
final ConsumerAcknowledgment acknowledgment = new ConsumerAcknowledgment() {
@Override
public void acknowledge(long partition, int offset) {
System.out.println("消息已确认：分区=" + partition + "，偏移量=" + offset);
}
};
// 消费消息并手动确认
while (true) {
ConsumerRecords records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord record : records) {
System.out.printf("消费消息：主题=%s，分区=%d，偏移量=%d，值=%s%n", record.topic(), record.partition(), record.offset(), record.value());
// 处理消息...
// 确认消息
acknowledgment.acknowledge(record.partition(), record.offset());
}
}
在这个示例中，我们创建了一个手动确认的处理器acknowledgment，并在消费消息后调用其acknowledge方法来发送确认信号。