深入理解RocketMQ--事务消息

刺骨的言语ヽ痛彻心扉 2022-12-28 09:11 242阅读 0赞

消息队列RocketMQ版提供的分布式事务消息适用于所有对数据最终一致性有强需求的场景。本文介绍消息队列RocketMQ版事务消息的概念、优势、典型场景、交互流程以及使用过程中的注意事项。

## 概念介绍 ##

*  事务消息：消息队列RocketMQ版提供类似X或Open XA的分布式事务功能，通过消息队列RocketMQ版事务消息能达到分布式事务的最终一致。
 *  半事务消息：暂不能投递的消息，发送方已经成功地将消息发送到了消息队列RocketMQ版服务端，但是服务端未收到生产者对该消息的二次确认，此时该消息被标记成“暂不能投递”状态，处于该种状态下的消息即半事务消息。
 *  消息回查：由于网络闪断、生产者应用重启等原因，导致某条事务消息的二次确认丢失，消息队列RocketMQ版服务端通过扫描发现某条消息长期处于“半事务消息”时，需要主动向消息生产者询问该消息的最终状态（Commit或是Rollback），该询问过程即消息回查。

## 分布式事务消息的优势 ##

消息队列RocketMQ版分布式事务消息不仅可以实现应用之间的解耦，又能保证数据的最终一致性。同时，传统的大事务可以被拆分为小事务，不仅能提升效率，还不会因为某一个关联应用的不可用导致整体回滚，从而最大限度保证核心系统的可用性。在极端情况下，如果关联的某一个应用始终无法处理成功，也只需对当前应用进行补偿或数据订正处理，而无需对整体业务进行回滚。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3k1MzI3OTgxMTM_size_16_color_FFFFFF_t_70][]

## 典型场景 ##

在淘宝购物车下单时，涉及到购物车系统和交易系统，这两个系统之间的数据最终一致性可以通过分布式事务消息的异步处理实现。在这种场景下，交易系统是最为核心的系统，需要最大限度地保证下单成功。而购物车系统只需要订阅消息队列RocketMQ版的交易订单消息，做相应的业务处理，即可保证最终的数据一致性。

## 交互流程 ##

事务消息交互流程如下图所示。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3k1MzI3OTgxMTM_size_16_color_FFFFFF_t_70 1][]

事务消息发送步骤如下：

1.  发送方将半事务消息发送至消息队列RocketMQ版服务端。
2.  消息队列RocketMQ版服务端将消息持久化成功之后，向发送方返回Ack确认消息已经发送成功，此时消息为半事务消息。
3.  发送方开始执行本地事务逻辑。
4.  发送方根据本地事务执行结果向服务端提交二次确认（Commit或是Rollback），服务端收到Commit状态则将半事务消息标记为可投递，订阅方最终将收到该消息；服务端收到Rollback状态则删除半事务消息，订阅方将不会接受该消息。

事务消息回查步骤如下：

1.  在断网或者是应用重启的特殊情况下，上述步骤4提交的二次确认最终未到达服务端，经过固定时间后服务端将对该消息发起消息回查。
2.  发送方收到消息回查后，需要检查对应消息的本地事务执行的最终结果。
3.  发送方根据检查得到的本地事务的最终状态再次提交二次确认，服务端仍按照步骤4对半事务消息进行操作。

## 注意事项 ##

1.  事务消息的Group ID不能与其他类型消息的Group ID共用。与其他类型的消息不同，事务消息有回查机制，回查时消息队列RocketMQ版服务端会根据Group ID去查询客户端。
2.  通过`ONSFactory.createTransactionProducer`创建事务消息的Producer时必须指定`LocalTransactionChecker`的实现类，处理异常情况下事务消息的回查。
3.  事务消息发送完成本地事务后，可在`execute`方法中返回以下三种状态：
    
     *  `TransactionStatus.CommitTransaction`：提交事务，允许订阅方消费该消息。
     *  `TransactionStatus.RollbackTransaction`：回滚事务，消息将被丢弃不允许消费。
     *  `TransactionStatus.Unknow`：暂时无法判断状态，等待固定时间以后消息队列RocketMQ版服务端向发送方进行消息回查。
4.  可通过以下方式给每条消息设定第一次消息回查的最快时间：
    
        Message message = new Message();
        // 在消息属性中添加第一次消息回查的最快时间，单位秒。例如，以下设置实际第一次回查时间为120秒~125秒之间message.putUserProperties(PropertyKeyConst.CheckImmunityTimeInSeconds,"120");
        // 以上方式只确定事务消息的第一次回查的最快时间，实际回查时间向后浮动0秒~5秒；如第一次回查后事务仍未提交，后续每隔5秒回查一次

# #

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3k1MzI3OTgxMTM_size_16_color_FFFFFF_t_70]: /images/20221120/3d146bad0c324284b30c1ba0db6ffacc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3k1MzI3OTgxMTM_size_16_color_FFFFFF_t_70 1]: /images/20221120/389fc81ef21c4ebb9ff9579b9ff70c6f.png