kafka偏移量保存到mysql里_Kafka 新版消费者 API（二）：提交偏移量

1. 自动提交

最简单的提交方式是让消费者自动提交偏移量。如果 enable.auto.commit 被设为 true，那么每过 5s，消费者会自动把从 poll() 方法接收到的最大偏移量提交上去。提交时间间隔由 auto.commit.interval.ms 控制，默认值是5s。消费者每次获取新数据时都会先把上一次poll()方法返回的最大偏移量提交上去。

可能造成的问题：数据重复读

假设我们仍然使用默认的 5s 提交时间间隔，在最近一次提交之后的 3s 发生了再均衡，再均衡之后，消费者从最后一次提交的偏移量位置开始读取消息。这个时候偏移量已经落后了 3s，所以在这 3s内到达的消息会被重复处理。可以通过修改提交时间间隔来更频繁地提交偏移量，减小可能出现重复消息的时间窗，不过这种情况是无法完全避免的。

2. 手动提交

(1) 同步提交

// 把auto.commit.offset设为false，让应用程序决定何时提交偏移量

props.put("auto.commit.offset", false);

try{

while(true) {

ConsumerRecords records = consumer.poll(1000);

for(ConsumerRecord record : records) {

// 假设把记录内容打印出来就算处理完毕

System.out.println("value = " + record.value() + ", topic = " + record.topic() +

", partition = " + record.partition() + ", offset = " + record.offset());

}

try{

// 只要没有发生不可恢复的错误，commitSync() 方法会一直尝试直至提交成功

// 如果提交失败，我们也只能把异常记录到错误日志里

consumer.commitSync();

}catch(CommitFailedException e) {

System.err.println("commit failed!" + e.getMessage());

}

}

}finally {

consumer.close();

}

(2) 异步提交

手动提交有一个不足之处，在 broker 对提交请求作出回应之前，应用程序会一直阻塞，这样会限制应用程序的吞吐量。我们可以通过降低提交频率来提升吞吐量，但如果发生了再均衡，会增加重复消息的数量。

这个时候可以使用异步提交，只管发送提交请求，无需等待 broker 的响应。

// 把auto.commit.offset设为false，让应用程序决定何时提交偏移量

props.put("auto.commit.offset", false);

try{

while(true) {

ConsumerRecords records = consumer.poll(1000);

for(ConsumerRecord record : records) {

System.out.println("value = " + record.value() + ", topic = " + record.topic() +

", partition = " + record.partition() + ", offset = " + record.offset());

}

// 提交最后一个偏移量，然后继续做其他事情。

consumer.commitAsync();

}

}finally {

consumer.close();

}

在成功提交或碰到无法恢复的错误之前，commitSync()会一直重试，但是commitAsync()不会，这也是commitAsync()不好的一个地方。它之所以不进行重试，是因为在它收到服务器响应的时候，可能有一个更大的偏移量已经提交成功。

假设我们发出一个请求用于提交偏移量2000，这个时候发生了短暂的通信问题，服务器收不到请求，自然也不会作出任何响应。与此同时，我们处理了另外一批消息，并成功提交了偏移量3000。如果commitAsync()重新尝试提交偏移量2000，它有可能在偏移量3000之后提交成功。这个时候如果发生再均衡，就会出现重复消息。

commitAsync()也支持回调，在broker作出响应时会执行回调：

// 把auto.commit.offset设为false，让应用程序决定何时提交偏移量

props.put("auto.commit.offset", false);

try {

while (true) {

ConsumerRecords records = consumer.poll(1000);

for (ConsumerRecord record : records) {

System.out.println("value = " + record.value() + ", topic = " + record.topic() + ", partition = "

+ record.partition() + ", offset = " + record.offset());

}

consumer.commitAsync(new OffsetCommitCallback() {

@Override

public void onComplete(Map offsets, Exception exception) {

if(offsets != null) {

System.out.println("commit offset successful!");

}

if(exception != null) {

System.out.println("commit offset fail!" + exception.getMessage());

}

}

});

}

} finally {

consumer.close();

}

可以在回调中重试失败的提交，以下为思路：

使用一个单调递增的序列号来维护异步提交的顺序。在每次提交偏移量之后或在回调里提交偏移量时递增序列号。在进行重试前，先检查回调的序列号和即将提交的偏移量是否相等，如果相等，说明没有新的提交，那么可以安全地进行重试。如果序列号比较大，说明有一个新的提交已经发送出去了，应该停止重试。

(3) 同步和异步组合提交

一般情况下，针对偶尔出现的提交失败，不进行重试不会有太大问题，因为如果提交失败是因为临时问题导致的，那么后续的提交总会有成功的。但如果这是发生在关闭消费者或再均衡前的最后一次提交，就要确保能够提交成功。

try {

while (true) {

ConsumerRecords records = consumer.poll(1000);

for (ConsumerRecord record : records) {

System.out.println("value = " + record.value() + ", topic = " + record.topic() + ", partition = "

+ record.partition() + ", offset = " + record.offset());

}

// 如果一切正常，我们使用 commitAsync() 方法来提交

// 这样速度更快，而且即使这次提交失败，下一次提交很可能会成功

consumer.commitAsync();

}

}catch (Exception e) {

e.printStackTrace();

}finally {

try {

// 使用 commitSync() 方法会一直重试，直到提交成功或发生无法恢复的错误

// 确保关闭消费者之前成功提交了偏移量

consumer.commitSync();

}finally {

consumer.close();

}

}

(4) 提交特定的偏移量

不管是自动提交还是使用commitAsync()或者commitSync()来提交偏移量，提交的都是 poll() 方法返回的那批数据的最大偏移量，想要自定义在什么时候提交偏移量可以这么做：

Map currentOffsets = new HashMap<>();

int count = 0;

......

try {

while (true) {

ConsumerRecords records = consumer.poll(1000);

for (ConsumerRecord record : records) {

System.out.println("value = " + record.value() + ", topic = " + record.topic() + ", partition = "

+ record.partition() + ", offset = " + record.offset());

currentOffsets.put(new TopicPartition(record.topic(), record.partition()),

new OffsetAndMetadata(record.offset() + 1, "no metadata"));

if (count % 1000 == 0) {

// 这里调用的是 commitAsync()，不过调用 commitSync() 也是完全可以的

// 当然，在提交特定偏移量时，仍然要处理可能发生的错误

consumer.commitAsync(currentOffsets, null);

}

count++;

}

}

}finally {

consumer.close();

}

3. 分区再均衡监听器

消费者在退出和进行分区再均衡之前，应该做一些正确的事情：

提交最后一个已处理记录的偏移量(必须做)

根据之前处理数据的业务不同，你可能还需要关闭数据库连接池、清空缓存等

程序如何能得知集群要进行"分区再均衡"了？消费者 API 提供了再均衡监听器，以下程序可以做到 kafka 消费数据的 Exactly Once 语义：

package com.bonc.rdpe.kafka110.consumer;

import java.util.Collection;

import java.util.Collections;

import java.util.HashMap;

import java.util.Map;

import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;

import org.apache.kafka.clients.consumer.ConsumerRecord;

import org.apache.kafka.clients.consumer.ConsumerRecords;

import org.apache.kafka.clients.consumer.KafkaConsumer;

import org.apache.kafka.clients.consumer.OffsetAndMetadata;

import org.apache.kafka.common.TopicPartition;

/**

* @Title RebalanceListenerConsumer.java

* @Description 再均衡监听器

* @Author YangYunhe

* @Date 2018-06-27 17:35:05

*/

public class RebalanceListenerConsumer {

public static void main(String[] args) {

Map currentOffsets = new HashMap<>();

Properties props = new Properties();

props.put("bootstrap.servers", "192.168.42.89:9092,192.168.42.89:9093,192.168.42.89:9094");

// 把auto.commit.offset设为false，让应用程序决定何时提交偏移量

props.put("auto.commit.offset", false);

props.put("group.id", "dev3-yangyunhe-group001");

props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

KafkaConsumer consumer = new KafkaConsumer<>(props);

consumer.subscribe(Collections.singletonList("dev3-yangyunhe-topic001"), new ConsumerRebalanceListener() {

/*

* 再均衡开始之前和消费者停止读取消息之后被调用

* 如果在这里提交偏移量，下一个接管分区的消费者就知道该从哪里开始读取了

*/

@Override

public void onPartitionsRevoked(Collection partitions) {

// 如果发生再均衡，我们要在即将失去分区所有权时提交偏移量

// 要注意，提交的是最近处理过的偏移量，而不是批次中还在处理的最后一个偏移量

System.out.println("Lost partitions in rebalance. Committing current offsets:" + currentOffsets);

consumer.commitSync(currentOffsets);

}

/*

* 在重新分配分区之后和新的消费者开始读取消息之前被调用

*/

@Override

public void onPartitionsAssigned(Collection partitions) {

long committedOffset = -1;

for(TopicPartition topicPartition : partitions) {

// 获取该分区已经消费的偏移量

committedOffset = consumer.committed(topicPartition).offset();

// 重置偏移量到上一次提交的偏移量的下一个位置处开始消费

consumer.seek(topicPartition, committedOffset + 1);

}

}

});

try {

while (true) {

ConsumerRecords records = consumer.poll(1000);

for (ConsumerRecord record : records) {

System.out.println("value = " + record.value() + ", topic = " + record.topic() + ", partition = "

+ record.partition() + ", offset = " + record.offset());

currentOffsets.put(new TopicPartition(record.topic(), record.partition()),

new OffsetAndMetadata(record.offset() + 1, "no metadata"));

}

consumer.commitAsync(currentOffsets, null);

}

} catch (Exception e) {

e.printStackTrace();

} finally {

try{

consumer.commitSync(currentOffsets);

} catch (Exception e) {

e.printStackTrace();

} finally {

consumer.close();

System.out.println("Closed consumer successfully!");

}

}

}

}

当然你也可以选择再均衡后从头开始消费：

consumer.subscribe(Collections.singletonList("dev3-yangyunhe-topic001"), new ConsumerRebalanceListener() {

@Override

public void onPartitionsRevoked(Collection partitions) {

System.out.println("starting partitions rebalance...");

}

@Override

public void onPartitionsAssigned(Collection partitions) {

consumer.seekToBeginning(partitions);

}

});

以上代码与 props.put("auto.offset.reset", "earliest");是等效的。

设置从最新消息开始消费：

consumer.subscribe(Collections.singletonList("dev3-yangyunhe-topic001"), new ConsumerRebalanceListener() {

@Override

public void onPartitionsRevoked(Collection partitions) {

System.out.println("starting partitions rebalance...");

}

@Override

public void onPartitionsAssigned(Collection partitions) {

consumer.seekToEnd(partitions);

}

});

以上代码与props.put("auto.offset.reset", "latest");等效。

4. 涉及到数据库的 Exactly Once 语义的实现思路

当处理 Kafka 中的数据涉及到数据库时，那么即使每处理一条数据提交一次偏移量，也可以造成数据重复处理或者丢失数据，看以下为伪代码：

Map currentOffsets = new HashMap<>();

......

while (true) {

ConsumerRecords records = consumer.poll(100);

for (ConsumerRecord record : records) {

currentOffsets.put(new TopicPartition(record.topic(), record.partition()),

new OffsetAndMetadata(record.offset() + 1);

// 处理数据

processRecord(record);

// 把数据存储到数据库中

storeRecordInDB(record);

// 提交偏移量

consumer.commitAsync(currentOffsets);

}

}

假设把数据存储到数据库后，没有来得及提交偏移量程序就因某种原因挂掉了，那么程序再次启动后就会重复处理数据，数据库中会有重复的数据。

如果把存储到数据库和提交偏移量在一个原子操作里完成，就可以避免这样的问题，但数据存到数据库，偏移量保存到kafka是无法实现原子操作的，而如果把数据存储到数据库中，偏移量也存储到数据库中，这样就可以利用数据库的事务来把这两个操作设为一个原子操作，同时结合再均衡监听器就可以实现 Exactly Once 语义，以下为伪代码：

consumer.subscribe(Collections topics, new ConsumerRebalanceListener() {

@Override

public void onPartitionsRevoked(Collection partitions) {

// 发生分区再均衡之前，提交事务

commitDBTransaction();

}

@Override

public void onPartitionsAssigned(Collection partitions) {

// 再均衡之后，从数据库获得消费偏移量

for(TopicPartition topicPartition : partitions) {

consumer.seek(topicPartition, getOffsetFromDB(topicPartition));

}

}

});

/**

* 消费之前调用一次 poll()，让消费者加入到消费组中，并获取分配的分区

* 然后马上调用 seek() 方法定位分区的偏移量

* seek() 设置消费偏移量，设置的偏移量是从数据库读出来的，说明本次设置的偏移量已经被处理过

* 下一次调用 poll() 就会在本次设置的偏移量上加1，开始处理没有处理过的数据

* 如果seek()发生错误，比如偏移量不存在，则会抛出异常

*/

consumer.poll(0);

for(TopicPartition topicPartition : consumer.assignment()) {

consumer.seek(topicPartition, getOffsetFromDB(topicPartition));

}

while (true) {

ConsumerRecords records = consumer.poll(1000);

for (ConsumerRecord record : records) {

// 处理数据

processRecord(record);

// 把数据存储到数据库中

storeRecordInDB(record);

// 把偏移量存储到数据库中

storeOffsetInDB(record.topic(), record.partition(), record.offset());

}

// 以上3步为一个事务，提交事务，这里在每个批次末尾提交一次事务，是为了提高性能

commitDBTransaction();

}

把偏移量和记录保存到用一个外部系统来实现 Exactly Once 有很多方法，但核心思想都是：结合 ConsumerRebalanceListener 和 seek() 方法来确保能够及时保存偏移量，并保证消费者总是能够从正确的位置开始读取消息。

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/300523.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

我居然手写了Spring框架

我居然手写了Spring框架

手写完了刚参加工作那会接触java还是用的struct的时代&#xff0c;后面在SSH火爆时代的时候我转战.net,多年之后公司转java技术栈已经是Spring的天下&#xff0c;源码嚼了很多遍于是很想尝试把这套东西用在.net平台上。社区有个Spring.net项目已经多年不维护了&#xff0c;而且…
阅读更多...
下班以后看什么，决定你人生的高度

下班以后看什么，决定你人生的高度

全世界只有3.14 % 的人关注了青少年数学之旅王小波说&#xff1a;我活在世上&#xff0c;无非想要明白些道理&#xff0c;遇见些有趣的事&#xff0c;倘能如我所愿&#xff0c;我的一生就算成功。你的圈子将决定你的人生。每一位对事物都有着独特的态度让你成为一个有趣的人。今…
阅读更多...
Mahout的taste推荐系统里的几种Recommender分析

Mahout的taste推荐系统里的几种Recommender分析

Taste简介 看自:http://blog.csdn.net/zhoubl668/article/details/13297583Mahout 是apache下的一个java语言的开源大数据机器学习项目&#xff0c;与其他机器学习项目不同的是&#xff0c;它的算法多数是mapreduce方式写的&#xff0c;可以在hadoop上运行&#xff0c;并行化处…
阅读更多...
[2007最后一博]Url地址重写，利用HttpHander手工编译页面并按需生成静态HTML文件...

[2007最后一博]Url地址重写，利用HttpHander手工编译页面并按需生成静态HTML文件...

很多朋友可能都讨论过ASP.NET中生成HTML的方法了&#xff0c;有按照模板用IO方法写文件 有在404错误页面内生成HTML的&#xff0c;有在Render内保存页面输出到HTML文件的。 今天我发一个配合Url重写利用HttpHander手工编译.aspx页面方法。 HTML文件的方法&#xff0c;可按需、“…
阅读更多...
found.000文件夹的问题

found.000文件夹的问题

found.000文件夹的问题问&#xff1a;我的电脑有的时候在C盘或D盘的根目录下有个名为found.000的文件夹&#xff0c;里面有一些后缀名为CHK的文件。在c:windows下有很多以fff开头的怪文件&#xff0c;而且大小全部为0字节。请问这些是什么文件&#xff1f;能否将它们删除&#…
阅读更多...
bytecode java_Java 字节码解读

bytecode java_Java 字节码解读

一、源代码public classPeople {privateString name;private intage;}二、利用Javap 反编译查看字节码结构信息javap -v People.class结果如下Classfile /D:/work/byte-code/src/People.classLast modified2020-12-9; size 240bytesMD5 checksum 54b8c1ad94a9c9cf5074fd8520454…
阅读更多...
他狂骗五千万美元消失17年...却被一个纪录片导演锲而不舍的追到了镜头前！...

他狂骗五千万美元消失17年...却被一个纪录片导演锲而不舍的追到了镜头前！...

全世界只有3.14 % 的人关注了青少年数学之旅他&#xff0c;曾经是个做啥都无师自通的天才。他&#xff0c;在众多的行业里&#xff0c;莫名进入到了艺术品行业&#xff0c;没过多久&#xff0c;他靠着惊人的自学能力&#xff0c;成了艺术圈里所有人津津乐道的画商。他&#xff…
阅读更多...
使用JDBC驱动程序访问SQL Server 2000数据库（实例）

使用JDBC驱动程序访问SQL Server 2000数据库（实例）

<!-- 使用JDBC驱动程序访问SQL Server 2000数据库 --><%String drivername"com.microsoft.jdbc.sqlserver.SQLServerDriver";//sql server 2000for jdbc 驱动程序名称String url"jdbc:microsoft:sqlserver://localhost:1433;DatabaseName学籍管理"…
阅读更多...
DotNetMagic 5.4.0破解

DotNetMagic 5.4.0破解

就不做多介绍&#xff0c;具体看我以前的文章&#xff0c;请勿用于商业用途&#xff0c;破解仅仅为学习研究之用&#xff0c;本人对此不承担任何法律责任&#xff0c;谢谢配合&#xff01;下载转载于:https://www.cnblogs.com/anqincmt/archive/2007/12/31/1021756.html
阅读更多...
合肥.NET俱乐部第二期技术沙龙活动预告

合肥.NET俱乐部第二期技术沙龙活动预告

各位亲爱的.NET从业和爱好者们&#xff1a; 大家好&#xff0c;自从19年举办.NET俱乐部第一期技术沙龙后&#xff0c;.NET在开源以及跨平台的加持下继续飞速发展&#xff0c;各种开源项目不断涌现&#xff0c;各种社区活动持续开展&#xff0c;.NET的生态逐步丰富&#xff0…
阅读更多...
你见过扇贝游泳吗？ | 今日趣图

你见过扇贝游泳吗？ | 今日趣图

全世界只有3.14 % 的人关注了青少年数学之旅正常颈椎VS颈椎病患者的颈椎丁香医生小学入学测试题你知道答案吗&#xff1f;图源网络为什么我的假期这么短&#xff1f;文丁香医生五子棋必胜秘籍图源网络那些女生称呼的真相图沙县小吃你有见过扇贝游泳吗小迷妹神吐槽
阅读更多...
Regulator的下载地址

Regulator的下载地址

都说Regulator“是一种很有特色的工具&#xff0c;能够使生成和测试正则表达式变得很容易”&#xff0c;想从网上下载试一下&#xff0c;不料全是清一色的 http://royo.is-a-geek.com/regulator &#xff08;不要试了&#xff0c;打不开的&#xff09;&#xff0c;转贴人真是不…
阅读更多...
java抓取新闻_【图片】【抓取】Java获取各大网站新闻【java吧】_百度贴吧

java抓取新闻_【图片】【抓取】Java获取各大网站新闻【java吧】_百度贴吧

该楼层疑似违规已被系统折叠 隐藏此楼查看此楼下面上源码&#xff0c;用一小时做的&#xff0c;还有些bugpackage Getnews;import java.io.*;import java.net.*;import com.itextpdf.text.List;public class GetNews {/*technology**** */static URL urlnull;staticInputStream…
阅读更多...
c#結合正則表達式驗證輸入的字符串

c#結合正則表達式驗證輸入的字符串

using System.Text.RegularExpressions; #region 驗證輸入的字符串 /// <summary> /// 判斷輸入的字符類型　 /// </summary> /// <param name"_value">輸入的字串</param> /// <param name"_kind">要驗…
阅读更多...
实名羡慕，国内这些厂.NET薪资高的吓人！

实名羡慕，国内这些厂.NET薪资高的吓人！

小米招C#工程师&#xff0c;要求WPF自动化设备&#xff0c;20~40k * 14薪&#xff1b;小鹏招WPF/.NET/PLC&#xff0c;25~40k年终&#xff1b;特斯拉招自动化控制16~25k * 15薪&#xff0c;华为应届硕士C#运动控制&#xff0c;就已经17~25k *14薪了。这些招聘让人眼热&#xff…
阅读更多...
修车工在生命最后，才知道自己的儿子是如今世界首富！

修车工在生命最后，才知道自己的儿子是如今世界首富！

全世界只有3.14 % 的人关注了青少年数学之旅在美国亚利桑那州凤凰城&#xff0c;曾经住着一个叫Ted Jorgensen的自行车修理工&#xff0c;他开了一家自行车店&#xff0c;平时默默的卖车修车&#xff0c;日子过的平平淡淡。在凤凰城郊区&#xff0c;他和二婚妻子住在这么一间不…
阅读更多...
没有情人的情人节

没有情人的情人节

我想和她一生一世。神却说&#xff1a;你只能爱她七天&#xff0c;我说&#xff1a;星期一到星期天。神&#xff1a;只四天呢&#xff1f;我&#xff1a;春天夏天秋天冬天。神&#xff1a;只三天呢? 我&#xff1a;今天明天后天。神&#xff1a;只两天&#xff1f;我&#xff…
阅读更多...
Sublime Text 2 中运行 PHP

Sublime Text 2 中运行 PHP

2019独角兽企业重金招聘Python工程师标准>>> Sublime Text 2 has the concept of build systems. This basically means that if you are editing a Python file then you can run the Python interpreter on the source file your are editing and see the output …
阅读更多...
理解 Delphi 的类(七) - 认识类的多态

理解 Delphi 的类(七) - 认识类的多态

什么是多态? 我的理解就是: 同样一个方法, 在不同的对象里会有不同的实现, 仅此而已.unit Unit1;interfaceusesWindows, Messages, SysUtils, Variants, Classes, Graphics, Controls, Forms,Dialogs, StdCtrls;typeTForm1 class(TForm)Button1: TButton;Button2: TButton;Bu…
阅读更多...
python退出帮助系统help应该使用exit_python--help - tesion

python退出帮助系统help应该使用exit_python--help - tesion

python help学习python的过程中&#xff0c;难免遇到一些生疏的东西&#xff0c;为此需要参看相关的文档说明。Linux下众所周知有man可以查找系统的命令帮助页&#xff1b;对于python也提供了自己的帮助文档的2种方式&#xff1a;方式一&#xff1a;启动python解释器&#xff0…
阅读更多...
最新文章

Copyright @ 2022~2023