当前位置： 首页 >> Kafka >> 手动提交offset

---

手动提交offset

2020-07-04 15:08:52 星期六  阅读：1385

---

虽然自动提交 offset 十分简介便利，但由于其是基于时间提交的，开发人员难以把握 offset 提交的时机。因此 Kafka 还提供了手动提交 offset 的API。 手动提交 offset 的方法有两种: **`commitSync(同步提交)`** **`commitAsync(异步提交)`** 两者的相同点：都会将本次 poll 的一批数据最高的偏移量提交; 两者的不同点：commitSync 阻塞当前线程，一直到提交成功，并且会自动失败重试(由不可控因素导致， 也会出现提交失败);而 commitAsync 则没有失败重试机制，故有可能提交失败。 手动提交的问题：无论是同步提交还是异步提交 offset，都有可能会造成数据的漏消费或者重复消费。先提交offset 后消费，有可能造成数据的漏消费;而先消费后提交 offset，有可能会造成数据的重复消费。所以`对于特别严谨的场景，可以自定义维护offset的存储（维护难度较大）`，可以将消费者的offset存放到MySQL（不存储到zk或者broker集群），利用MySQL的事务功能将消费者的offset存储与消费逻辑进行绑定。 ####手动同步提交 同步提交的offset有失败重试机制，所以更加可靠。具体实现方式为： 1、在构造消费者配置信息的时候关闭自动提交功能 ``` // 关闭自动提交 properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); ``` 2、在消费完一个批次之后调用同步提交的方法手动提交 ``` // 同步提交，当前线程会阻塞直到offset提交成功 consumer.commitSync(); ``` 提交位置如下 ``` while (true) { ConsumerRecords<Integer, String> records = consumer.poll(Duration.ofSeconds(100)); for (ConsumerRecord<Integer, String> record : records) { System.out.println(record.key() + " " + record.value() + "-> offset:" + record.offset()); } // 同步提交，当前线程会阻塞直到offset提交成功 consumer.commitSync(); } ``` ####手动异步提交 虽然同步提交offset更可靠一些，但是由于会阻塞当前线程，直到提交成功。因此吞吐量会受到很大的影响。因此在生产环境中，会选用异步提交offset的方式。具体的实现方式为： 1、在构造消费者配置信息的时候关闭自动提交功能 ``` // 关闭自动提交 properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); ``` 2、在消费完一个批次之后调用异步提交的方法提交 ``` //异步提交 consumer.commitAsync(new OffsetCommitCallback() { @Override public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception e) { if (e != null) { System.out.println("Commit failed for " + offsets); } } }); ``` 提交位置如下所示 ``` while (true) { ConsumerRecords<Integer, String> records = consumer.poll(Duration.ofSeconds(100)); for (ConsumerRecord<Integer, String> record : records) { System.out.println(record.key() + " " + record.value() + "-> offset:" + record.offset()); } //异步提交 consumer.commitAsync(new OffsetCommitCallback() { @Override public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception e) { if (e != null) { System.out.println("Commit failed for " + offsets); } } }); } ```