200字范文 > Kafka：Consumer手动提交offset

Kafka：Consumer手动提交offset

时间：2023-02-09 16:22:47

在上一篇博客中介绍了使用Consumer订阅多个Topic或者多个Partition：

Kafka：Consumer订阅

在上一篇博客的测试样例中，Consumer都是自动提交offset，这是通过下面的配置来生效：

// 开启offset自动提交properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");// 消费者offset自动提交到Kafka的频率（以毫秒为单位）properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");

如果Consumer手动提交offset，则需要将ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG设置为false。

测试代码

pom.xml：

<?xml version="1.0" encoding="UTF-8"?><project xmlns="/POM/4.0.0"xmlns:xsi="/2001/XMLSchema-instance"xsi:schemaLocation="/POM/4.0.0 /xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>com.kaven</groupId><artifactId>kafka</artifactId><version>1.0-SNAPSHOT</version><properties><piler.source>8</piler.source><piler.target>8</piler.target></properties><dependencies><dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>3.0.0</version></dependency></dependencies></project>

创建Topic：

package com.kaven.kafka.admin;import org.apache.kafka.clients.admin.*;import org.mon.KafkaFuture;import java.util.Collections;import java.util.Map;import java.util.Properties;import java.util.concurrent.CountDownLatch;import java.util.concurrent.ExecutionException;public class Admin {// 基于Kafka服务地址与请求超时时间来创建AdminClient实例private static final AdminClient adminClient = Admin.getAdminClient("192.168.1.9:9092,192.168.1.9:9093,192.168.1.9:9094","40000");public static void main(String[] args) throws InterruptedException, ExecutionException {Admin admin = new Admin();// 创建Topic，Topic名称为topic1，分区数为1，复制因子为1admin.createTopic("topic1", 1, (short) 1);// 创建Topic，Topic名称为topic2，分区数为2，复制因子为1admin.createTopic("topic2", 2, (short) 1);// 创建Topic，Topic名称为topic3，分区数为2，复制因子为1admin.createTopic("topic3", 2, (short) 1);Thread.sleep(10000);}public static AdminClient getAdminClient(String address, String requestTimeoutMS) {Properties properties = new Properties();properties.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, address);properties.setProperty(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, requestTimeoutMS);return AdminClient.create(properties);}public void createTopic(String name, int numPartitions, short replicationFactor) throws InterruptedException {CountDownLatch latch = new CountDownLatch(1);CreateTopicsResult topics = adminClient.createTopics(Collections.singleton(new NewTopic(name, numPartitions, replicationFactor)));Map<String, KafkaFuture<Void>> values = topics.values();values.forEach((name__, future) -> {future.whenComplete((a, throwable) -> {if(throwable != null) {System.out.println(throwable.getMessage());}System.out.println(name__);latch.countDown();});});latch.await();}}

Producer发布消息：

package com.kaven.kafka.producer;import org.apache.kafka.clients.producer.*;import java.util.Properties;import java.util.concurrent.ExecutionException;public class ProducerTest {public static void main(String[] args) throws ExecutionException, InterruptedException {send("topic1");send("topic2");send("topic3");}public static void send(String name) throws ExecutionException, InterruptedException {Producer<String, String> producer = ProducerTest.createProducer();for (int i = 0; i < 7; i++) {ProducerRecord<String, String> producerRecord = new ProducerRecord<>(name,"key-" + i,"value-" + i);// 异步发送并回调producer.send(producerRecord, (metadata, exception) -> {if(exception == null) {System.out.printf("topic: %s, partition: %s, offset: %s\n", name, metadata.partition(), metadata.offset());}else {exception.printStackTrace();}});}// 要关闭Producer实例producer.close();}public static Producer<String, String> createProducer() {// Producer的配置Properties properties = new Properties();// 服务地址properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.9:9092,192.168.1.9:9093,192.168.1.9:9094");// KEY的序列化器类properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.mon.serialization.StringSerializer");// VALUE的序列化器类properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.mon.serialization.StringSerializer");return new KafkaProducer<>(properties);}}

不提交offset

Consumer订阅程序：

package com.kaven.kafka.consumer;import org.apache.kafka.clients.consumer.*;import java.time.Duration;import java.util.*;public class ConsumerTest {public static void main(String[] args) {commit(Arrays.asList("topic1", "topic2", "topic3"));}public static void commit(List<String> topicList) {KafkaConsumer<String, String> consumer = createConsumer();consumer.subscribe(topicList);while (true) {ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(10000));records.forEach((record) -> {System.out.printf("topic: %s, partition: %s, offset: %s, key: %s, value: %s\n",record.topic(), record.partition(), record.offset(), record.key(), record.value());});}}public static KafkaConsumer<String, String> createConsumer() {// Consumer的配置Properties properties = new Properties();// 服务地址properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.9:9092,192.168.1.9:9093,192.168.1.9:9094");// 组ID，用于标识此消费者所属的消费者组properties.put(ConsumerConfig.GROUP_ID_CONFIG, "kaven-test");// 关闭offset自动提交properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");// 消费者offset自动提交到Kafka的频率（以毫秒为单位）// properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");// KEY的反序列化器类properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.mon.serialization.StringDeserializer");// VALUE的反序列化器类properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.mon.serialization.StringDeserializer");return new KafkaConsumer<>(properties);}}

先创建Topic，然后运行Consumer订阅程序，目前还没有消息发布：

通过Producer发布消息到Topic上：

topic: topic1, partition: 0, offset: 56topic: topic1, partition: 0, offset: 57topic: topic1, partition: 0, offset: 58topic: topic1, partition: 0, offset: 59topic: topic1, partition: 0, offset: 60topic: topic1, partition: 0, offset: 61topic: topic1, partition: 0, offset: 62topic: topic2, partition: 0, offset: 32topic: topic2, partition: 0, offset: 33topic: topic2, partition: 0, offset: 34topic: topic2, partition: 0, offset: 35topic: topic2, partition: 1, offset: 24topic: topic2, partition: 1, offset: 25topic: topic2, partition: 1, offset: 26topic: topic3, partition: 1, offset: 24topic: topic3, partition: 1, offset: 25topic: topic3, partition: 1, offset: 26topic: topic3, partition: 0, offset: 32topic: topic3, partition: 0, offset: 33topic: topic3, partition: 0, offset: 34topic: topic3, partition: 0, offset: 35

此时Consumer就可以订阅到消息了，输出如下所示：

topic: topic2, partition: 1, offset: 24, key: key-0, value: value-0topic: topic2, partition: 1, offset: 25, key: key-3, value: value-3topic: topic2, partition: 1, offset: 26, key: key-4, value: value-4topic: topic1, partition: 0, offset: 56, key: key-0, value: value-0topic: topic1, partition: 0, offset: 57, key: key-1, value: value-1topic: topic1, partition: 0, offset: 58, key: key-2, value: value-2topic: topic1, partition: 0, offset: 59, key: key-3, value: value-3topic: topic1, partition: 0, offset: 60, key: key-4, value: value-4topic: topic1, partition: 0, offset: 61, key: key-5, value: value-5topic: topic1, partition: 0, offset: 62, key: key-6, value: value-6topic: topic2, partition: 0, offset: 32, key: key-1, value: value-1topic: topic2, partition: 0, offset: 33, key: key-2, value: value-2topic: topic2, partition: 0, offset: 34, key: key-5, value: value-5topic: topic2, partition: 0, offset: 35, key: key-6, value: value-6topic: topic3, partition: 1, offset: 24, key: key-0, value: value-0topic: topic3, partition: 1, offset: 25, key: key-3, value: value-3topic: topic3, partition: 1, offset: 26, key: key-4, value: value-4topic: topic3, partition: 0, offset: 32, key: key-1, value: value-1topic: topic3, partition: 0, offset: 33, key: key-2, value: value-2topic: topic3, partition: 0, offset: 34, key: key-5, value: value-5topic: topic3, partition: 0, offset: 35, key: key-6, value: value-6

因为上次消费没有提交offset，再次运行Consumer订阅程序，会重复消费消息：

topic: topic2, partition: 0, offset: 32, key: key-1, value: value-1topic: topic2, partition: 0, offset: 33, key: key-2, value: value-2topic: topic2, partition: 0, offset: 34, key: key-5, value: value-5topic: topic2, partition: 0, offset: 35, key: key-6, value: value-6topic: topic3, partition: 1, offset: 24, key: key-0, value: value-0topic: topic3, partition: 1, offset: 25, key: key-3, value: value-3topic: topic3, partition: 1, offset: 26, key: key-4, value: value-4topic: topic2, partition: 1, offset: 24, key: key-0, value: value-0topic: topic2, partition: 1, offset: 25, key: key-3, value: value-3topic: topic2, partition: 1, offset: 26, key: key-4, value: value-4topic: topic1, partition: 0, offset: 56, key: key-0, value: value-0topic: topic1, partition: 0, offset: 57, key: key-1, value: value-1topic: topic1, partition: 0, offset: 58, key: key-2, value: value-2topic: topic1, partition: 0, offset: 59, key: key-3, value: value-3topic: topic1, partition: 0, offset: 60, key: key-4, value: value-4topic: topic1, partition: 0, offset: 61, key: key-5, value: value-5topic: topic1, partition: 0, offset: 62, key: key-6, value: value-6topic: topic3, partition: 0, offset: 32, key: key-1, value: value-1topic: topic3, partition: 0, offset: 33, key: key-2, value: value-2topic: topic3, partition: 0, offset: 34, key: key-5, value: value-5topic: topic3, partition: 0, offset: 35, key: key-6, value: value-6

提交offset

修改commit方法：

public static void commit(List<String> topicList) {KafkaConsumer<String, String> consumer = createConsumer();consumer.subscribe(topicList);while (true) {ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(10000));records.forEach((record) -> {System.out.printf("topic: %s, partition: %s, offset: %s, key: %s, value: %s\n",record.topic(), record.partition(), record.offset(), record.key(), record.value());});mitSync();}}

因为上一次消费没有提交offset，运行Consumer订阅程序还是会消费上一次消费过的消息，但这次消费之后提交了offset（通过mitSync()）。

再次运行Consumer订阅程序，就不会消费上次消费过的消息了。

基于Partition提交offset

Consumer订阅程序：

package com.kaven.kafka.consumer;import org.apache.kafka.clients.consumer.*;import org.mon.TopicPartition;import java.time.Duration;import java.util.*;public class ConsumerTest {public static void main(String[] args) {commitWithPartition(Arrays.asList("topic1", "topic2", "topic3"));}public static void commitWithPartition(List<String> topicList) {KafkaConsumer<String, String> consumer = createConsumer();consumer.subscribe(topicList);while (true) {ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(10000));records.partitions().forEach((partition) -> {List<ConsumerRecord<String, String>> recordsWithPartition = records.records(partition);recordsWithPartition.forEach((record) -> {System.out.printf("topic: %s, partition: %s, offset: %s, key: %s, value: %s\n",record.topic(), record.partition(), record.offset(), record.key(), record.value());});// 为了测试效果，只提交分区1的offsetif(partition.partition() == 1) {// 更新offset，即上一次最大的offset + 1long newOffset = recordsWithPartition.get(recordsWithPartition.size() - 1).offset() + 1;Map<TopicPartition, OffsetAndMetadata> newOffsetMap = new HashMap<>();newOffsetMap.put(partition, new OffsetAndMetadata(newOffset));mitSync(newOffsetMap);}});}}public static KafkaConsumer<String, String> createConsumer() {// Consumer的配置Properties properties = new Properties();// 服务地址properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.9:9092,192.168.1.9:9093,192.168.1.9:9094");// 组ID，用于标识此消费者所属的消费者组properties.put(ConsumerConfig.GROUP_ID_CONFIG, "kaven-test");// 关闭offset自动提交properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");// 消费者offset自动提交到Kafka的频率（以毫秒为单位）// properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");// KEY的反序列化器类properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.mon.serialization.StringDeserializer");// VALUE的反序列化器类properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.mon.serialization.StringDeserializer");return new KafkaConsumer<>(properties);}}

运行Consumer订阅程序，然后使用Producer发布消息：

topic: topic1, partition: 0, offset: 63topic: topic1, partition: 0, offset: 64topic: topic1, partition: 0, offset: 65topic: topic1, partition: 0, offset: 66topic: topic1, partition: 0, offset: 67topic: topic1, partition: 0, offset: 68topic: topic1, partition: 0, offset: 69topic: topic2, partition: 1, offset: 27topic: topic2, partition: 1, offset: 28topic: topic2, partition: 1, offset: 29topic: topic2, partition: 0, offset: 36topic: topic2, partition: 0, offset: 37topic: topic2, partition: 0, offset: 38topic: topic2, partition: 0, offset: 39topic: topic3, partition: 1, offset: 27topic: topic3, partition: 1, offset: 28topic: topic3, partition: 1, offset: 29topic: topic3, partition: 0, offset: 36topic: topic3, partition: 0, offset: 37topic: topic3, partition: 0, offset: 38topic: topic3, partition: 0, offset: 39

此时Consumer就可以订阅到消息了，输出如下所示：

topic: topic2, partition: 0, offset: 36, key: key-1, value: value-1topic: topic2, partition: 0, offset: 37, key: key-2, value: value-2topic: topic2, partition: 0, offset: 38, key: key-5, value: value-5topic: topic2, partition: 0, offset: 39, key: key-6, value: value-6topic: topic3, partition: 1, offset: 27, key: key-0, value: value-0topic: topic3, partition: 1, offset: 28, key: key-3, value: value-3topic: topic3, partition: 1, offset: 29, key: key-4, value: value-4topic: topic2, partition: 1, offset: 27, key: key-0, value: value-0topic: topic2, partition: 1, offset: 28, key: key-3, value: value-3topic: topic2, partition: 1, offset: 29, key: key-4, value: value-4topic: topic1, partition: 0, offset: 63, key: key-0, value: value-0topic: topic1, partition: 0, offset: 64, key: key-1, value: value-1topic: topic1, partition: 0, offset: 65, key: key-2, value: value-2topic: topic1, partition: 0, offset: 66, key: key-3, value: value-3topic: topic1, partition: 0, offset: 67, key: key-4, value: value-4topic: topic1, partition: 0, offset: 68, key: key-5, value: value-5topic: topic1, partition: 0, offset: 69, key: key-6, value: value-6topic: topic3, partition: 0, offset: 36, key: key-1, value: value-1topic: topic3, partition: 0, offset: 37, key: key-2, value: value-2topic: topic3, partition: 0, offset: 38, key: key-5, value: value-5topic: topic3, partition: 0, offset: 39, key: key-6, value: value-6

再运行Consumer订阅程序，Consumer还可以消费到上次没有提交offset的消息（都是分区0中的消息）：

topic: topic1, partition: 0, offset: 63, key: key-0, value: value-0topic: topic1, partition: 0, offset: 64, key: key-1, value: value-1topic: topic1, partition: 0, offset: 65, key: key-2, value: value-2topic: topic1, partition: 0, offset: 66, key: key-3, value: value-3topic: topic1, partition: 0, offset: 67, key: key-4, value: value-4topic: topic1, partition: 0, offset: 68, key: key-5, value: value-5topic: topic1, partition: 0, offset: 69, key: key-6, value: value-6topic: topic3, partition: 0, offset: 36, key: key-1, value: value-1topic: topic3, partition: 0, offset: 37, key: key-2, value: value-2topic: topic3, partition: 0, offset: 38, key: key-5, value: value-5topic: topic3, partition: 0, offset: 39, key: key-6, value: value-6topic: topic2, partition: 0, offset: 36, key: key-1, value: value-1topic: topic2, partition: 0, offset: 37, key: key-2, value: value-2topic: topic2, partition: 0, offset: 38, key: key-5, value: value-5topic: topic2, partition: 0, offset: 39, key: key-6, value: value-6

Consumer手动提交offset就介绍到这里，如果博主有说错的地方或者大家有不同的见解，欢迎大家评论补充。

本内容不代表本网观点和政治立场，如有侵犯你的权益请联系我们处理。

网友评论

网友评论仅供其表达个人看法，并不表明网站立场。