在Kafka中,消费者通过提交偏移量来跟踪它们已经处理过的消息。默认情况下,消费者会自动提交偏移量,但也可以配置为手动提交。以下是手动提交偏移量的步骤:
- 创建一个
Properties
对象,用于存储消费者的配置信息。
Properties props = new Properties();
- 配置消费者的
bootstrap.servers
、group.id
和key.deserializer
等基本属性。
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");
- 创建一个
KafkaConsumer
实例,并将配置信息传递给它。
KafkaConsumerconsumer = new KafkaConsumer<>(props);
- 使用
subscribe
方法订阅一个或多个主题。
consumer.subscribe(Arrays.asList("my-topic"));
- 在处理消息的循环中,使用
poll
方法获取消息。
while (true) { ConsumerRecordsrecords = consumer.poll(Duration.ofMillis(100)); for (ConsumerRecord record : records) { // 处理消息的逻辑 System.out.printf("offset = %d, key = %s, value = https://www.yisu.com/ask/%s%n", record.offset(), record.key(), record.value()); } }
- 在处理完消息后,使用
commitSync
方法手动提交偏移量。这将提交当前处理的所有分区的偏移量。
consumer.commitSync();
注意:在生产环境中,建议使用commitAsync
方法异步提交偏移量,以提高性能。但请注意,异步提交可能会导致数据丢失,因此需要确保正确处理异常情况。
- 当不再需要消费者时,调用
close
方法关闭消费者。
consumer.close();
通过以上步骤,你可以实现Kafka消费者的手动提交偏移量。