kafka生产者幂等与事务

目录

前言:

幂等

事务

总结: 

参考资料 


前言:

Kafka 消息交付可靠性保障以及精确处理一次语义的实现。

所谓的消息交付可靠性保障,是指 Kafka 对 Producer 和 Consumer 要处理的消息提供什么样的承诺。常见的承诺有以下三种:

  • 最多一次(at most once):消息可能会丢失,但绝不会被重复发送。
  • 至少一次(at least once):消息不会丢失,但有可能被重复发送。
  • 精确一次(exactly once):消息不会丢失,也不会被重复发送。

目前,Kafka 默认提供的交付可靠性保障是第二种,即至少一次。 

   即只有 Broker 成功“提交”消息且 Producer 接到 Broker 的应答才会认为该消息成功发送。不过倘若消息成功“提交”,但 Broker 的应答没有成功发送回 Producer 端(比如网络出现瞬时抖动),那么 Producer 就无法确定消息是否真的提交成功了。因此,它只能选择重试,也就是再次发送相同的消息。这就是 Kafka 默认提供至少一次可靠性保障的原因,不过这会导致消息重复发送。

大部分用户还是希望消息只会被交付一次,这样的话,消息既不会丢失,也不会被重复处理。或者说,即使 Producer 端重复发送了相同的消息,Broker 端也能做到自动去重。在下游 Consumer 看来,消息依然只有一条。 

      Kafka 是怎么做到精确一次的呢?简单来说,这是通过两种机制:幂等性(Idempotence)和事务(Transaction)。

幂等

   “幂等”这个词原是数学领域中的概念,指的是某些操作或函数能够被执行多次,但每次得到的结果都是不变的。

    幂等性有很多好处,其最大的优势在于我们可以安全地重试任何幂等性操作,反正它们也不会破坏我们的系统状态。如果是非幂等性操作,我们还需要担心某些操作执行多次对状态的影响,但对于幂等性操作而言,我们根本无需担心此事。

        在 Kafka 中,Producer 默认不是幂等性的,但我们可以创建幂等性 Producer。它其实是 0.11.0.0 版本引入的新功能。在此之前,Kafka 向分区发送数据时,可能会出现同一条消息被发送了多次,导致消息重复的情况。在 0.11 之后,指定 Producer 幂等性的方法很简单,仅需要设置一个参数即可,即 props.put(“enable.idempotence”, ture),或 props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true)。

       enable.idempotence 被设置成 true 后,Producer 自动升级成幂等性 Producer,其他所有的代码逻辑都不需要改变。Kafka 自动帮你做消息的重复去重。底层具体的原理很简单,就是经典的用空间去换时间的优化思路,即在 Broker 端多保存一些字段。当 Producer 发送了具有相同字段值的消息后,Broker 能够自动知晓这些消息已经重复了,于是可以在后台默默地把它们“丢弃”掉。当然,实际的实现原理并没有这么简单,但你大致可以这么理解。

      看上去,幂等性 Producer 的功能很酷,使用起来也很简单,仅仅设置一个参数就能保证消息不重复了,但实际上,我们必须要了解幂等性 Producer 的作用范围。

      首先,它只能保证单分区上的幂等性,即一个幂等性 Producer 能够保证某个主题的一个分区上不出现重复消息,它无法实现多个分区的幂等性。次,它只能实现单会话上的幂等性,不能实现跨会话的幂等性。这里的会话,你可以理解为 Producer 进程的一次运行。当你重启了 Producer 进程之后,这种幂等性保证就丧失了。

      那么你可能会问,如果我想实现多分区以及多会话上的消息无重复,应该怎么做呢?答案就是事务(transaction)或者依赖事务型 Producer。这也是幂等性 Producer 和事务型 Producer 的最大区别!

事务

 Kafka 的事务概念类似于我们熟知的数据库提供的事务。在数据库领域,事务提供的安全性保障是经典的 ACID,即原子性(Atomicity)、一致性 (Consistency)、隔离性 (Isolation) 和持久性 (Durability)。

各大主流数据库厂商都比较统一。所谓的 read committed,指的是当读取数据库时,你只能看到已提交的数据,即无脏读。同时,当写入数据库时,你也只能覆盖掉已提交的数据,即无脏写。

Kafka 自 0.11 版本开始也提供了对事务的支持,目前主要是在 read committed 隔离级别上做事情。它能保证多条消息原子性地写入到目标分区,同时也能保证 Consumer 只能看到事务成功提交的消息。下面我们就来看看 Kafka 中的事务型 Producer。

事务型 Producer 能够保证将消息原子性地写入到多个分区中。这批消息要么全部写入成功,要么全部失败。另外,事务型 Producer 也不惧进程的重启。Producer 重启回来后,Kafka 依然保证它们发送消息的精确一次处理。

设置事务型 Producer 的方法也很简单,满足两个要求即可:

  • 和幂等性 Producer 一样,开启 enable.idempotence = true。
  • 设置 Producer 端参数 transactional. id。最好为其设置一个有意义的名字

此外,你还需要在 Producer 代码中做一些调整,如这段代码所示: 

producer.initTransactions();
try {producer.beginTransaction();producer.send(record1);producer.send(record2);producer.commitTransaction();
} catch (KafkaException e) {producer.abortTransaction();
}

nitTransaction、beginTransaction、commitTransaction 和 abortTransaction,它们分别对应事务的初始化、事务开始、事务提交以及事务终止。

这段代码能够保证 Record1 和 Record2 被当作一个事务统一提交到 Kafka,要么它们全部提交成功,要么全部写入失败。实际上即使写入失败,Kafka 也会把它们写入到底层的日志中,也就是说 Consumer 还是会看到这些消息。因此在 Consumer 端,读取事务型 Producer 发送的消息也是需要一些变更的。修改起来也很简单,设置 isolation.level 参数的值即可。当前这个参数有两个取值:

  1. read_uncommitted:这是默认值,表明 Consumer 能够读取到 Kafka 写入的任何消息,不论事务型 Producer 提交事务还是终止事务,其写入的消息都可以读取。很显然,如果你用了事务型 Producer,那么对应的 Consumer 就不要使用这个值。
  2. read_committed:表明 Consumer 只会读取事务型 Producer 成功提交事务写入的消息。当然了,它也能看到非事务型 Producer 写入的所有消息。

总结: 

幂等性 Producer 和事务型 Producer 都是 Kafka 社区力图为 Kafka 实现精确一次处理语义所提供的工具,只是它们的作用范围是不同的。幂等性 Producer 只能保证单分区、单会话上的消息幂等性;而事务能够保证跨分区、跨会话间的幂等性。从交付语义上来看,自然是事务型 Producer 能做的更多。

不过,切记天下没有免费的午餐。比起幂等性 Producer,事务型 Producer 的性能要更差,在实际使用过程中,我们需要仔细评估引入事务的开销,切不可无脑地启用事务。

参考资料 

14 | 幂等生产者和事务生产者是一回事吗?-极客时间

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

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

相关文章

No view found for id 0x7f0901c3 for fragment解决以及线上bug排查技巧

情景再现 开发这么久,不知道你们是否也经历过这样的情况,测试或者用户,反馈app闪退,结果你自己打开开发工具,去调试,一切正常,然后闪退还是存在,只是在开发环境中不能重现。这种情况…

boost下的asio异步高并发tcp服务器搭建

C 网络编程 asio 使用总结 - 知乎 (zhihu.com) 基于Boost::asio的多线程异步TCP服务器&#xff0c;实现了io_service线程池&#xff0c;测试了1万左右的并发访问&#xff0c;读写无压力_boost asio支持最大并发_E404的博客-CSDN博客 单线程 server.cpp #include <cstdlib&g…

【ARM 嵌入式 编译系列 11.1 -- GCC __attribute__((aligned(x)))详细介绍】

文章目录 __attribute__((aligned(x)))详细介绍其它对齐方式 上篇文章&#xff1a;ARM 嵌入式 编译系列 11 – GCC attribute&#xff08;(packed)&#xff09;详细介绍 attribute((aligned(x)))详细介绍 __attribute__((aligned(x))) 是 GCC 编译器的一个特性&#xff0c;它可…

SpringBoot代理访问本地静态资源400 404

SpringBoot代理访问静态资源400 404 背景&#xff1a;pdf文件上传到linux服务器上&#xff0c;使用SpringBoot代理访问问题&#xff1a;访问过程中可能会出现400、404问题 前提&#xff1a;保证有文件&#xff0c;并且文件路径正确 SpringBoot如何配置静态资源代理&#xff0…

Flutter实现倒计时功能,秒数转时分秒,然后倒计时

Flutter实现倒计时功能 发布时间&#xff1a;2023/05/12 本文实例为大家分享了Flutter实现倒计时功能的具体代码&#xff0c;供大家参考&#xff0c;具体内容如下 有一个需求&#xff0c;需要在页面进行显示倒计时&#xff0c;倒计时结束后&#xff0c;做相应的逻辑处理。 实…

Antd的日期选择器中文化配置

当你使用antd的日期选择器后&#xff0c;你会发现日期什么都是英文的&#xff1a;即便你已经在项目中配置了中文化&#xff1a; 我确实已经配置了中文化&#xff1a; 但是为啥没生效&#xff1f;官网回答&#xff1a;FAQ - Ant Design dayjs中文网&#xff1a; 安装 | Day…

零拷贝详解

1、在没有DMA技术之前的I/O过程是这样的&#xff1a; CPU发出对应的指令给磁盘控制器&#xff0c;然后返回磁盘控制器收到指令后&#xff0c;于是就开始准备数据&#xff0c;会把数据放入到磁盘控制器的内部缓冲区&#xff0c;然后产生中断CPU收到中断信号后&#xff0c;停下手…

华为OD机试-5键键盘的输出

题目描述 【5键键盘的输出】有一个特殊的 5键键盘&#xff0c;上面有 a,ctrl-c,ctrl-x,ctrl-v,ctrl-a五个键。 a键在屏幕上输出一个字母 a; ctrl-c将当前选择的字母复制到剪贴板; ctrl-x将当前选择的 字母复制到剪贴板&#xff0c;并清空选择的字母; ctrl-v将当前剪贴板里的字母…

HTML是什么?

HTML是什么&#xff1f; 超文本标记语言&#xff08;英语&#xff1a;HyperText Markup Language&#xff0c;简称&#xff1a;HTML&#xff09;是一种用于创建网页的标准标记语言。 您可以使用 HTML 来建立自己的 WEB 站点&#xff0c;HTML 运行在浏览器上&#xff0c;由浏览器…

【业务功能篇63】Springboot聊聊 过滤器和拦截器

过滤器的场景&#xff1a;过滤器通常用于对数据或资源进行筛选、修改或转换的场景。例如&#xff0c;在一个电子商务网站中&#xff0c;用户进行商品搜索时&#xff0c;你可以使用过滤器来过滤特定的商品类别、价格范围或其他条件&#xff0c;以便用户仅看到符合筛选条件的结果…

人工智能时代的科学探索 | 《自然》评述

人工智能(AI)正越来越多地融入科学发现&#xff0c;以增强和加速研究&#xff0c;帮助科学家提出假设、设计实验、收集和解释大型数据集&#xff0c;并获得仅靠传统科学方法可能无法实现的洞察力。 过去十年间&#xff0c;AI取得了巨大的突破。其中就包括自监督学习和几何深度学…

手机的发展历史

目录 一.人类的通信方式变化 二.手机对人类通信的影响 三.手机的发展过程 四.手机对现代人的影响 一.人类的通信方式变化 人类通信方式的变化是一个非常广泛和复杂的话题&#xff0c;随着技术的进步和社会的发展&#xff0c;人类通信方式发生了许多重大的变化。下面是一些主…

go mod使用最新提交依赖

例如一个项目在其中依赖了 github.com/linuxsuren/go-fake-runtime v0.0.1 go.mod内容&#xff1a; github.com/linuxsuren/go-fake-runtime v0.0.1 修改了github.com/linuxsuren/go-fake-runtime代码&#xff0c;存在一个最新的commit hash值为25fa814c6232e545f5bce03bd…

【opencv】指定宽或高按比例缩放图片 拼接图片

指定宽或高按比例缩放图片 import cv2def resize_by_ratio(image, widthNone, heightNone, intercv2.INTER_AREA):img_new_size None(h, w) image.shape[:2] # 获得高度和宽度if width is None and height is None: # 如果输入的宽度和高度都为空return image # 直接返回原图…

应用程序运行报错:First section must be [net] or [network]:No such file or directory

应用程序报错环境&#xff1a; 在linux下&#xff0c;调用darknet训练的模型&#xff0c;报错&#xff1a;First section must be [net] or [network]:No such file or directory&#xff0c;并提示&#xff1a;"./src/utils.c:256: error: Assertion 0 failed." 如…

百日筑基篇——Pandas学习三(pyhton入门八)

百日筑基篇——Pandas学习三&#xff08;pyhton入门八&#xff09; 文章目录 前言一、数据排序二、字符串处理三、数据合并方法1. merge方法2. concat方法 四、分组数据统计五、数据重塑1. stack2. pivot 总结 前言 上一篇文章介绍了一下pandas库中的一些函数&#xff0c;而本…

MySQL数据类型

文章目录 MySQL数据类型1. 数据类型分类2. 数值类型2.1 tinyint类型2.2 bit类型2.3 小数类型2.3.1 float2.3.2 decimal 2.4 字符串类型2.4.1 char2.4.2 varchar2.4.3 char和varchar比较 2.5 日期和时间类型2.6 enum和set MySQL数据类型 1. 数据类型分类 红色标注是我主要讲解…

【QT】 QFileQFileInfo文件操作

很高兴在雪易的CSDN遇见你 &#xff0c;给你糖糖 欢迎大家加入雪易社区-CSDN社区云 前言 本文分享QT对文件的操作技术&#xff0c;希望对各位小伙伴有所帮助&#xff01; 感谢各位小伙伴的点赞关注&#xff0c;小易会继续努力分享&#xff0c;一起进步&#xff01; 你的点…

linux中profile.d和profile的区别

profile.d在profile中加载 profile文件 PATH"/bin:/sbin:/usr/bin:/usr/sbin:/opt/bin:/opt/scripts:/soc/bin:/soc/scripts" LD_LIBRARY_PATH"/usr/local/lib:/usr/lib:/opt/lib:/soc/lib" export SSL_LDPATH/usr/local/lib/ export ZLIB_LDPATH/usr/lo…

appium默认60秒关闭应用的问题

问题&#xff1a;appium默认启动一个应用的session过期时间是60秒到时间会自动停了刚启动的应用&#xff0c;工作台打印&#xff1a;info: [debug] We shut down because no new commands came in的日志 分析&#xff1a;--command-timeout 60 The default command timeout fo…