消息队列高手课
李玥
京东零售技术架构部资深架构师
立即订阅
8426 人已学习
课程目录
已完结 41 讲
0/4登录后,你可以任选4讲全文学习。
课前必读 (2讲)
开篇词 | 优秀的程序员,你的技术栈中不能只有“增删改查”
免费
预习 | 怎样更好地学习这门课?
基础篇 (8讲)
01 | 为什么需要消息队列?
02 | 该如何选择消息队列?
03 | 消息模型:主题和队列有什么区别?
04 | 如何利用事务消息实现分布式事务?
05 | 如何确保消息不会丢失?
06 | 如何处理消费过程中的重复消息?
07 | 消息积压了该如何处理?
08 | 答疑解惑(一) : 网关如何接收服务端的秒杀结果?
进阶篇 (21讲)
09 | 学习开源代码该如何入手?
10 | 如何使用异步设计提升系统性能?
11 | 如何实现高性能的异步网络传输?
12 | 序列化与反序列化:如何通过网络传输结构化的数据?
13 | 传输协议:应用程序之间对话的语言
14 | 内存管理:如何避免内存溢出和频繁的垃圾回收?
加餐 | JMQ的Broker是如何异步处理消息的?
15 | Kafka如何实现高性能IO?
16 | 缓存策略:如何使用缓存来减少磁盘IO?
17 | 如何正确使用锁保护共享数据,协调异步线程?
18 | 如何用硬件同步原语(CAS)替代锁?
19 | 数据压缩:时间换空间的游戏
20 | RocketMQ Producer源码分析:消息生产的实现过程
21 | Kafka Consumer源码分析:消息消费的实现过程
22 | Kafka和RocketMQ的消息复制实现的差异点在哪?
23 | RocketMQ客户端如何在集群中找到正确的节点?
24 | Kafka的协调服务ZooKeeper:实现分布式系统的“瑞士军刀”
25 | RocketMQ与Kafka中如何实现事务?
26 | MQTT协议:如何支持海量的在线IoT设备?
27 | Pulsar的存储计算分离设计:全新的消息队列设计思路
28 | 答疑解惑(二):我的100元哪儿去了?
案例篇 (7讲)
29 | 流计算与消息(一):通过Flink理解流计算的原理
30 | 流计算与消息(二):在流计算中使用Kafka链接计算任务
31 | 动手实现一个简单的RPC框架(一):原理和程序的结构
32 | 动手实现一个简单的RPC框架(二):通信与序列化
33 | 动手实现一个简单的RPC框架(三):客户端
34 | 动手实现一个简单的RPC框架(四):服务端
35 | 答疑解惑(三):主流消息队列都是如何存储消息的?
测试篇 (2讲)
期中测试丨10个消息队列热点问题自测
免费
期末测试 | 消息队列100分试卷等你来挑战!
结束语 (1讲)
结束语 | 程序员如何构建知识体系?
消息队列高手课
登录|注册

04 | 如何利用事务消息实现分布式事务?

李玥 2019-07-30
你好,我是李玥,今天我们来聊一聊消息和事务。
一说起事务,你可能自然会联想到数据库。的确,我们日常使用事务的场景,绝大部分都是在操作数据库的时候。像 MySQL、Oracle 这些主流的关系型数据库,也都提供了完整的事务实现。那消息队列为什么也需要事务呢?
其实很多场景下,我们“发消息”这个过程,目的往往是通知另外一个系统或者模块去更新数据,消息队列中的“事务”,主要解决的是消息生产者和消息消费者的数据一致性问题。
依然拿我们熟悉的电商来举个例子。一般来说,用户在电商 APP 上购物时,先把商品加到购物车里,然后几件商品一起下单,最后支付,完成购物流程,就可以愉快地等待收货了。
这个过程中有一个需要用到消息队列的步骤,订单系统创建订单后,发消息给购物车系统,将已下单的商品从购物车中删除。因为从购物车删除已下单商品这个步骤,并不是用户下单支付这个主要流程中必需的步骤,使用消息队列来异步清理购物车是更加合理的设计。
对于订单系统来说,它创建订单的过程中实际上执行了 2 个步骤的操作:
在订单库中插入一条订单数据,创建订单;
发消息给消息队列,消息的内容就是刚刚创建的订单。
购物车系统订阅相应的主题,接收订单创建的消息,然后清理购物车,在购物车中删除订单中的商品。
取消
完成
0/1000字
划线
笔记
复制
© 版权归极客邦科技所有,未经许可不得传播售卖。 页面已增加防盗追踪,如有侵权极客邦将依法追究其法律责任。
该试读文章来自付费专栏《消息队列高手课》,如需阅读全部文章,
请订阅文章所属专栏。
立即订阅
登录 后留言

精选留言(112)

  • 一步
    对于上面订单的例子,为什么不等待订单创建成功再向消息队列发送订单数据呢?这样串行的话,对性能影响应该也不大,也不用考虑订单创建失败而发送消息的情况了。

    作者回复: 考虑这样一种情况:订单创建成功了,还没来得及发消息,这个节点突然断电了。

    2019-07-30
    19
    60
  • 微微一笑
    老师您好,下面是我对思考题的一些看法:
    A:本地事物的操作1,与往消息队列中生产消息的操作2,是两个分离的操作,不符合对原子性的定义;
    C:由于操作消息队列属于异步操作,在数据一致性上,只能保证数据的最终一致性。若对于时效性要求很高的系统来说,事物消息不是数据一致的;但对于时效性要求不高的系统来说,他就是数据一致的。我认为,用不同的业务视角来看问题,会有不同的答案;
    I:隔离性上,由于事物消息是分两步操作的,本地事物提交后,别的事物消息就已经可以看到提交的消息了。所以,不符合隔离性的定义;
    D:持久性上,rocketMq上支持事物的反查机制,但我不太清楚“半消息”是存储在磁盘中,还是内存里。若存储在磁盘中,那就支持持久性,即使事物消息提交后,发生服务突然宕机也不受影响;若存储在内存中,则无法保证持久性。
    以上是我的理解,望老师指点~

    作者回复: 这个总结非常到位,给你点赞!

    2019-07-30
    3
    39
  • ly
    实现订单下单场景:
    1. 首先通过producer.sendMessageInTransaction()方法发送一个半消息给MQ.
    2. 此时会在TransactionListener中的executeLocalTransaction()方法阻塞,然后在这个方法里面进行订单创建并提交本地事务,如果commit成功,则返回COMMIT状态,否则是ROLLBACK状态,如果正常返回COMMIT或者ROLLBACK的话,不会存在第3步的反查情况。
    3. 如果上面的本地事务提交成功以后,此节点突然断电,那么checkLocalTransaction()反查方法就会在某个时候被MQ调用,此方法会根据消息中的订单号去数据库确认订单是否存在,存在就返回COMMIT状态,否则是ROLLBACK状态。
    4. 购物车在另外一个项目中,反正只要收到MQ的消息就将本次订单的商品从购物车中删除即可。

    以上是通过代码的进行步骤写的,老师看有没有什么问题。

    作者回复: 非常好,完全正确!

    2019-07-30
    3
    17
  • oscarwin
    先开启本地事务,然后创建订单,订单创建成功后再发消息,根据发消息是否成功来决定提交还是回滚本地事务。这样不需要事务消息也能解决这个场景的问题了?还是说我考虑的不够全面。

    作者回复: 如果本地事务提交失败,已经发出去的消息是无法撤回的,会导致数据不一致。

    2019-07-31
    4
    14
  • linqw
    使用rocketmq实现分布式事务的理解和疑问,老师有空帮忙解答下哦
    1、rocketmq实现分布式事务,使用的是两阶段提交,和mysql写redo log和binlog日志的两阶段提交类似,以上面订单的为例,提交订单消息到mq中,等待mq回复ack,消息提交成功,但是此时的消息对消费组不可见,即half消息,此阶段像mysql的引擎层写redo log的prepare阶段,执行本地事务,执行本地事务成功,此阶段像mysql的service层写binlog的阶段,写binlog成功,最后提交或者回滚队列事务,rocketmq为了防止commit和rollback超时或者失败,采取回查的补偿机制,回查次数默认15次(感觉这个会不会导致服务超时了),超过会rollback,有点像mysql宕机重启根据redo log中的xid找binlog的xid事务,如果binlog日志也已经写成功,mysql这个事务也会提交,因为redo log和binlog这个事务都写完整。
    2、消息对消费者不可见,将其消息的主题topic和队列id修改为half topic,原先的主题和队列id也做为消息的属性,如果事务提交或者回滚会将其消息的队列改为原先的队列。rocketMq开启任务,从half topic中获取消息,调用其中的生产者的监听进行回查是否提交回滚。
    3、rocketmq采用commitlog存放消息,消费者使用consumeQueue二级索引从commitlog获取消息实体内容,不太理解Index File:索引文件?回查借助OP topic进行获取到Half消息进行后续的回查操作,感觉整体流程还是没有串通,老师能否帮忙解答下么?

    作者回复: indexFile的作用就是给commitlog做的索引,提升读取消息时的查询效率。

    另外,关于事务的实现流程,总结的很到位,你还有哪些具体的问题不清楚,可以继续留言提出来。

    2019-07-30
    5
    10
  • Calix
    这个半消息,和生活中的“交定金”有点类似。

    作者回复: 其实是交全款,不发货。

    2019-08-06
    1
    7
  • DC
    rocket mq 事务消息参考文档:https://rocketmq.apache.org/docs/transaction-example/

    作者回复: 👍👍👍

    2019-08-01
    6
  • 君莫笑
    老师,我回头重新看的时候看到这一章有一点疑问,消息队列的手动确认模式是可以保证分布式事务的最终一致性,那么如果生产者在处理完自己的业务之后将消息放入消息队列中(通过生产者确认方式可以确保消息送达Broker),然后消费者消费这个消息的时候出了问题,假设是消息体本身的原因导致消费该消息一定会抛出异常,这种情况下怎么通知生产者回滚该消息所处理的业务数据呢?

    作者回复: 这种情况下是没有办法回滚的,也不应该回滚。
    因为对于消息队列来说,它的一个重要功能就是解耦。
    消费者的任何行为,不应该影响生产者。

    对于你说的“坏消息”,反复消费都不能成功,有的MQ会把这种消息放到一个单独的特殊队列中,等着后续人工处理,避免卡死队列。

    2019-08-26
    1
    5
  • Geek
    老师,有几个问题没有太理解,可以解答一下么?
    1.kafka在commit/rollback的时候如果发送失败了就会抛出异常,会不会存在已经发送成功了但是超时了的情况呢,这个时候broker已经收到数据了。但是上游业务却回滚了
    2.RocketMq反查时有没有可能本地事务还没提交呢,导致broker取消了事务造成了不一致
    3.RocketMq在反查时如果订单服务异常了,导致broker取消了会不会导致事务造成了不一致
    谢谢老师

    作者回复: 第一个问题,我们后面还有专门的一节课来讲事务是如何实现的,这里面会有你想要的答案。

    第二第三个问题,RocketMQ给出的解决方案是,反查的结果返回的状态中,不仅有成功和失败,还有一个“不确定”的状态,意思就是“我现在不知道本地事务是不是成功了,将来它可能会成功,也可能会失败”,像你提的这两种情况,在实现反查接口的时候,都应该返回不确定的状态,RocketMQ在收到这个状态后,会定时多次进行反查,直到得到成功、失败的状态或者事务超时才结束。

    2019-08-21
    5
  • 芥末小龙
    我jiao着吧,如果先创建订单,当前服务由于不可抗拒因素不能正常工作了,没有给购物车系统发送消息,这种情况加就会出现 订单已经创建并且购物车没有清空的情况。然鹅发送半消息这种情况,可以通过定期查询事务的状态然后根据然后具体的业务回滚操作或者重新发送消息(保持业务的幂等性)。技术渣理解的有可能不到位 谅解

    作者回复: 到位。

    2019-07-30
    4
  • 朱海昆
    老师,如果消息队列不支持半消息,是否有其他的解决方案?我了解到一种解决方法是利用数据库的事务消息表的方案。把消息信息的快照和对业务数据的操作作为数据库事务操作数据库,操作成功后从数据库读取消息信息发送给broker,收到发送成功的回执后删除数据库中的消息快照。我个人觉得这种方案在不支持半消息的队列方案里也是一种选择,不知道您觉得这种实现方案有没有什么问题。

    作者回复: 如果有一个生产者和消费者都可以访问的,并且性能还不错数据库,肯定是使用这个数据库来实现事务比较好。

    大部分事务消息使用的场景是,没有这样的数据库的。或者由于设计、安全或者网络原因,生产者消费者不能共享数据库,或者是数据库的性能达不到要求。

    2019-07-30
    1
    4
  • yan
    如果订单ID是要创建完订单才会有的,那消息中就没有订单ID,那反查本地事务要根据什么查?

    作者回复: 所以几乎所有的类似系统都会事先生成订单ID,而不是在插入数据库的时候才生成。

    2019-09-26
    1
    3
  • Yize Li
    看了之前的一个留言 认为本地数据库和消息系统是两个系统所以违反了原子性 我是有些疑惑的。
    我认为 原子性破坏与否取决于是否存在数据库中订单成功但是在购物车中商品没有取消的情况 通过rocketmq的半消息模式是可以保证该情况不出现。所以原子性没有破坏 但是由于消息系统的异步性 导致我们可以观察到事物执行过程中或回滚中的中间状态 这意味着强一致性被破坏 只剩下了最终一致性

    作者回复: 是这样的。

    2019-09-22
    3
  • 佳明
    消费端做幂等处理来保障消息不会重复消费:1. 可以采用状态机的方式。2.消息数据唯一键+redis setnx来保障。3.本地消息表,要确保插入本地消息表和执行消息消费业务在同一事务里。
    2019-08-01
    3
  • 长期规划
    老师,kafka对提交消息队列事务失败时的处理方法,我感觉和不用事务没什么区别吧?如果不用事务,先完成数据库事务,再发消息,消失失败也是用户自己处理,比如重试。这跟kafka用分布式事务有什么区别吗

    作者回复: 把数据库事务放到Kafka事务中的好处是,在发消息和执行SQL阶段,无论哪个操作失败了,都可以自动回滚。
    只有“提交Kafka事务失败了”这个情况才需要手动处理。

    而要是不用Kafka事务,先执行数据库事务,再发普通消息,如果发消息失败就要手动回滚数据库。

    你要明白,提交事务只是设置一个状态,失败的概率要远远小于发消息的概率。

    虽然理论上这两种情况失败都需手动处理,但是失败的概率差别很大,所以实际上还是非常有用的。

    2019-09-18
    2
    2
  • 我瑟瑟的方法
    文中的订单购物车列子,第二步发送半消息第三步创建订单,这2个顺序反一下是不是也是等价的,即先创建订单在发送半消息

    作者回复: 可以的。

    2019-08-05
    2
  • Mark Yao
    对RocketMQ 的事务反查机制比较感兴趣。如果反查的服务器数据不一致,它是认为本地事务失败还是继续多次反查呢?

    作者回复: 你可以看一下这个反查接口的定义,它检查的是本地事务(在我们这个例子里面就是数据库事务)有没有执行成功,并不比较数据是否一致。

    2019-07-30
    2
  • 许童童
    老师讲得真好,真的是深入浅出,继续跟着老师学习!

    作者回复: 坚持下来就是对老师最好的支持。

    2019-07-30
    2
  • 高志强
    老师开启事物是基于什么实现,是关系型数据库还是kafka本身,如果提交失败,但发出去的消息已经撤不回来了...

    作者回复: 你可以继续学习这篇:25 | RocketMQ与Kafka中如何实现事务?

    2019-12-03
    1
  • 文正
    我们之前其实最古老的做法是,同步将待发消息存储在另一个消息表里面,这样就能保证了业务与本地消息在同一个事务中。然后启动一个任务专门去发消息。当然这样也有不好的地方。为了便于理解,完全可以这样逆向的来思考,将消息事务理解成另一个数据库事务。

    作者回复: 嗯 ,其实这两种方法原理是差不多的。

    2019-10-20
    1
收起评论
99+
返回
顶部