RocketMQ源码之事务消息！

Apache RocketMQ在4.3.0版中已经支持分布式事务消息。

  • RocketMQ采用了2PC的思想来实现了提交事务消息，同时增加一个补偿逻辑来处理二阶段超时或者失败的消息。

img

基本流程

第一阶段：

  • 发送 Message，Half Message，即半事务消息。
  • 此类型的 Message 是不会被 Consumer 消费。

第二阶段：如果半事务消息投递成功，则会开始执行本地事务。

分为如下三种 Case：

  • 本地事务执行成功：

    • 会向 Broker 发送 commit 消息，被 commit 过后的 Message 才能被 Consumer 消费到。

  • 本地事务执行失败：

    • 会向 Broker 发送 rollback 消息，Broker 则会将刚刚投递的半事务消息删除，从而保证上下游数据的一致性。

  • 如果 Producer 实例或者网络出现了问题，Producer 没能及时地将本地事务执行的结果通知 Broker。

  • Broker 会通过扫描发现某条 Message 长时间处于半事务消息状态。

  • Broker 会主动地向 Producer 询问此 Message 对应的事务状态。

基本设计

采用2PC两阶段设计:

img

将 Message 原本真实的 Topic 和 MessageQueue 进行备份。

  • 放入到PROPERTY_REAL_TOPIC、PROPERTY_REAL_QUEUE_ID中保存。

将消息投递到一个内部Topic中RMQ_SYS_TRANS_HALF_TOPIC，该队列专门存储事务消息。

所有的 Half Message 全部都写入到 queueId 为 0 的 MessageQueue。

因为一个 Topic 下只有 1 个 MessageQueue：

  • 这个 Topic 下的所有 Message 就是全局有序的，它们会按照先来后到的顺序被消费。

如果本地事务执行成功进行Commit，则将RMQ_SYS_TRANS_HALF_TOPIC 队列中的消息投递到真实的Topic中，供后续流程执行。

  • 并删除这条 Half Message ，但删除也是假删除，只是给 Message 打上一个删除的 Tag。

如果本地事务执行失败进行rollback，则直接删除这条 Half Message ，但删除也是假删除。

如果本地事务迟迟没有返回结果 (默认时间是6s)，则会触发事务回查机制

  • 执行回查之前需要校验检查次数是否到达了最大值（需要手动设置，没有默认值）。
  • 或者是当前 Half Message 存在是否超过了 Message 保存的上限，即 3天。
  • 如果满足上面条件中的一种Half Message 会被放进 TRANS_CHECK_MAX_TIME_TOPIC Topic 当中。
  • 一旦判定为需要执行事务回查逻辑，那么当前这条 Half Message 就算已经被消费了。
  • 在没达到最大的校验次数之前，都还需要将其投递到事务队列当中，以便下次重试时再次执行 Check 逻辑。
  • 如果回查成功则删除投递的 Half Message。