如何使用消息队列的事务消息

1 MQ事务的意义

“发消息”过程，往往是为通知另外一个系统更新数据，MQ的“事务”，主要解决消息生产者和消息消费者的数据一致性问题。

用户在电商APP上购物时

1. 先把商品加到购物车
2. 然后几件商品一起下单
3. 最后支付
4. 完成购物流程，就可以愉快地等待收货

该过程中有个需用MQ。 订单系统创建订单后，发消息给购物车模块，将已下单商品从购物车删除。 从购物车删除已下单商品步骤，并非用户下单支付这个主要流程的必需步骤，所以使用MQ异步清理购物车更合理。

订单模块创建订单的过程实际执行了俩操作：

1. 在订单DB插一条订单数据，用来创建订单
2. 发消息给MQ，消息内容即刚创建的订单

购物车模块订阅相应主题，接收订单创建的消息，然后清理购物车，在购物车中删除订单中的商品。

分布式下的这些步骤都有失败可能性，若不做处理，就可能导致订单数据与购物车数据不一致：

1. 创建了订单，没有清理购物车
2. 订单没创建成功，购物车里面的商品却被清了

• 因此问题 在任意步骤都可能失败时，要保证订单DB和购物车DB的数据一致性。

购物车系统收到订单创建成功消息清理购物车操作，只要成功执行购物车清理后再提交消费确认即可 如果失败，由于没有提交消费确认，MQ会自动重试。

问题关键点在订单系统，创建订单和发送消息不允许一个成功而另一个失败。 这就是事务问题。

分布式事务

单体关系型数据库都完整的实现ACID，但对分布式系统

• 严格实现ACID几乎不可能
• 或者说实现代价太大，大到无法接受

分布式系统在保证可用性和不严重牺牲性能的前提下，要实现数据一致性非常困难，所以出现很多“残血版”一致性，比如顺序一致性、最终一致性。

所以分布式事务更多是在分布式系统中事务的不完整实现。在不同场景有不同实现，都是通过一些妥协解决问题。

常见分布式事务实现有2PC、TCC和事务消息。 每种实现都有其特定的使用场景，也有各自问题，都不是完美方案。

事务消息适用场景

主要是那些需要异步更新数据，并且对数据实时性要求不高。 比如在创建订单后，如果出现短暂几秒，购物车商品没被及时清空，也不是完全不可接受，只要最终购物车的数据和订单数据保持一致。

MQ实现分布式事务

事务消息需要MQ提供相应功能才能实现，Kafka和RocketMQ都提供事务相关功能。

第二步发送半消息第三步创建订单，这2个顺序反一下是等价的，即先创建订单在发送半消息。

半消息并非消息内容不完整，包含的就是完整的消息内容。

• 和普通消息唯一区别 在事务提交前，对消费者，该消息不可见。这个半消息，不是和生活中的“交定金”有点类似。其实是交全款，不发货。

半消息发成功后，订单系统就可执行本地事务： 在订单库创建一条订单记录，并提交订单库的数据库事务。 然后根据本地事务执行结果决定提交或者回滚事务消息。

• 订单创建成功，提交事务消息，购物车系统即可消费到该消息，继续后续流程
• 订单创建失败，回滚事务消息，购物车系统不会收到该消息

这就基本实现“都成功/失败”的一致性要求。

但这实现过程，有个问题没有解决：如果在第4步提交事务消息时失败怎么办？ Kafka和RocketMQ给了不同解决方案。

• Kafka简单粗暴，直接抛异常，让用户自行处理 我们可在业务代码反复重试提交，直到提交成功 或者删除之前创建的订单进行补偿。

在我们这里例子里面，本地事务就是创建订单这个数据库事务。

若MQ不支持半消息，是否有其他的解决方案

利用数据库的事务消息表。 把消息信息的快照和对业务数据的操作作为数据库事务操作数据库，操作成功后从数据库读取消息信息发送给broker，收到发送成功的回执后删除数据库中的消息快照。我个人觉得这种方案在不支持半消息的队列方案里也是一种选择，不知道您觉得这种实现方案有没有什么问题。

如果有个生产者和消费者都可访问，并且性能还不错的数据库，肯定使用这个数据库实现事务较好。 然而大部分事务消息使用的场景是

• 没有这样的数据库
• 或由于设计、安全或者网络原因，生产者消费者不能共享数据库
• 或数据库的性能达不到要求

如果先创建订单，当前服务由于不可抗拒因素不能正常工作，没给购物车系统发送消息，这种情况加就会出现：订单已创建且购物车没有清空。 而发送半消息，可通过定期查询事务状态然后根据然后具体的业务回滚操作或者重新发送消息（保持业务的幂等性）。

消费端做幂等处理来保障消息不会重复消费

1. 可以采用状态机的方式
2. 消息数据唯一键+redis setnx来保障
3. 本地消息表，要确保插入本地消息表和执行消息消费业务在同一事务里

RocketMQ分布式事务

RocketMQ事务实现增加了事务反查机制来解决事务消息提交失败的问题。 如果Producer（即订单模块），在提交或回滚事务消息时发生网络异常，Broker没有收到提交或回滚请求，Broker会定期去Producer反查该事务对应的本地事务的状态，然后根据反查结果决定提交或者回滚该事务。

要支持事务反查机制，业务代码需实现一个反查本地事务状态的接口，告知RocketMQ本地事务是成功还是失败。

如果反查的服务器数据不一致，它是认为本地事务失败还是继续多次反查呢？

反查接口的定义，它检查的是本地事务（在我们这个例子里面就是数据库事务）有没有执行成功，并不比较数据是否一致。

该例中反查本地事务逻辑简单，只要根据消息中订单ID，在订单库中查询该订单是否存在，若订单存在则返回成功，否则返回失败。 RocketMQ会自动根据事务反查的结果提交或者回滚事务消息。

反查本地事务的实现并不依赖消息的发送方，即订单服务的某节点的任何数据。 这种情况下，即使发送事务消息的订单服务节点宕机，RocketMQ依然可通过其他订单服务节点执行反查，确保事务完整性。

RocketMQ事务消息流程图

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

若插入消息表成功后，消费方崩溃导致消费失败

因为消费失败，会自动重试，所以不会丢消息，但可能重复消费。

回查时间和次数怎么设置

如果发布者本地事务执行太久还没执行完，消息中心就来回查是不是有问题，所以应可以把发消息放本地事务的后面吧，另外次数定义也是经验值吧

反查一般是定一个事务超时时间，超时之前会不定期回查。

事务回查感觉需要特定业务自己实现，消息体里需要带回查的参数来判断本地事务结果

• 有无通用做法，比如发消息前产生唯一id写入数据库，后面回查用这个id来查事务结果就好了，而无需具体逻辑自己做 目前还没有通用的做法，需要业务方自己实现。

RocketMQ事务消息代码实现本案例

代码实现订单下单：

1. 首先通过producer.sendMessageInTransaction()方法发半消息给MQ

1. 此时会在TransactionListener中的executeLocalTransaction()方法阻塞，然后在这个方法里面进行订单创建并提交本地事务 如果commit成功，则返回COMMIT状态 否则是ROLLBACK状态，如果正常返回COMMIT或者ROLLBACK的话，不会存在第3步的反查情况。

1. 如果上面的本地事务提交成功以后，此节点突然断电，那么checkLocalTransaction()反查方法就会在某个时候被MQ调用，此方法会根据消息中的订单号去数据库确认订单是否存在，存在就返回COMMIT状态，否则是ROLLBACK状态。

1. 购物车在另一模块，只要收到MQ消息就将本次订单的商品从购物车中删除即可。

RocketMQ事务消息完整实现ACID了吗

• A:本地事务的操作1，与往MQ中生产消息的操作2，是两个分离操作，不符合原子性
• C:由于操作MQ属异步，在数据一致性上，只能保证最终一致性。 对时效性要求很高系统，事务消息并非数据一致 但对时效性要求不高系统，就是数据一致的 需要结合业务需要看问题
• I：由于事务消息分两步操作，本地事务提交后，别的事务消息就已经可看到提交的消息。所以，不符合隔离性
• D：rocketMq上支持事物的反查机制，但“半消息”是存储在磁盘中，还是内存呢？
  • 若存储在磁盘中，那就支持持久性，即使事物消息提交后，发生服务突然宕机也不受影响
  • 若存储在内存，则无法保证持久性

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这个事务都写完整。

消息对消费者不可见，将其消息的主题topic和队列id修改为half topic，原先的主题和队列id也做为消息的属性，如果事务提交或者回滚会将其消息的队列改为原先的队列。rocketMq开启任务，从half topic中获取消息，调用其中的生产者的监听进行回查是否提交回滚。

rocketmq采用commitlog存放消息，消费者使用consumeQueue二级索引从commitlog获取消息实体内容。 理解Index File：indexFile的作用就是给commitlog做的索引，提升读取消息时的查询效率。 回查借助OP topic进行获取到Half消息进行后续的回查操作。

总结

RocketMQ事务反查机制通过定期反查事务状态，来补偿提交事务消息可能出现的通信失败。 在Kafka的事务功能中，并没有类似的反查机制，需要用户自行去解决这个问题。

但不代表RocketMQ的事务功能比Kafka更好，只能说在该例场景，RocketMQ更适合。 Kafka对事务的定义、实现和适用场景，和RocketMQ有较大差异。

• 参考 https://rocketmq.apache.org/docs/transaction-example/