使用Rocket MQ事务消息达到数据最终一致性

使用Rocket MQ事务消息达到数据最终一致性

参考

1、 http://rocketmq.apache.org/rocketmq/the-design-of-transactional-message/ Rocket MQ事务消息的设计

2、https://rocketmq.apache.org/docs/transaction-example/ RocketMQ 事务消息的例子

0x00 前言

在分布式环境下，经常会有跨服务的事务需求，典型的例子如: 服务A 为账户服务，服务B为包月服务，在服务A扣钱成功之后，要在服务B上做增加包月时长的操作，需要保证数据的一致性。本文主要描写如何在分布式环境下通过Rocket MQ的事务消息保证数据的最终一致性。

注: 最终一致性是指在中间过程中，数据有可能不一致，但经过一段时间后(具体时长视网络延迟以及系统负载而定)，数据最终会保持一致。

0x01 Rocket MQ事务消息原理

Rocket MQ是阿里开发的一个分布式的开源消息队列组件，目前由Apache开源组织维护，最新版本是4.3.0，已经支持事务消息。

事务消息可以确保本地事务 与 发送消息 之间的原子性，相关概念:

1、Half(Prepare) Message

Producer已经把消息发送给Mq 服务器，但是Mq服务器尚未收到生产者的第二次Ack，这个时候消息会被标记为"temporarily undeliverable"，目前消息的状态为 Half Message。

2、Message Status Check

网络断开或者Producer应用重启会导致Mq服务器无法从Producer获取第二次ACK，当Mq服务器发现一个消息长时间处于 Half Message 状态时(默认为60S，可配置)，它会主动请求Producer，查询消息Id对应的最新状态(commit 或者 rollback)。

具体流程如下:

1、Producer 向Mq服务器 发送消息。

2、Mq服务器收到消息并持久化成功之后，会向 Producer确认首次ACK，此时消息处于 Half Message 状态，并未发送给对应的Consumer。

3、Producer 开始执行本地事务逻辑。

4、根据事务执行结果，Producer 向Mq服务器提交二次确认(commit 或者 rollback)。Mq Server 收到 Commit 状态则将半消息标记为可投递，Consumer 最终将收到该消息；MQ Server 收到 Rollback 状态则删除半消息，Consumer 将不会接受该消息。

5、在断网或者应用重启的情况下，二次ACK未成功的发给Mq Server，Mq Server会主动向 Producer 启动消息回查(Message Status Check)，

6、Producer 根据事务执行结果，对消息回查返回对应的结果。

7、Mq Server根据返回结果，决定继续投递消息或者丢弃消息(重复第4步操作)。

流程图如下:

事务消息流程.png

0x03 实例

针对上面案例，可以通过如下的方式使用事务消息。

Producer：

1、实现 TransactionListener 接口，在executeLocalTransaction方法里执行本地事务逻辑，在checkLocalTransaction方法里返回消息id对应的事务状态，用于Mq的消息回查。

2、通过 TransactionMQProducer构造事务消息并发送。

Consumer:

1、从Mq server获取到消息之后，即开始处理本地事务，处理成功后返回 CONSUME_SUCCESS。

2、处理失败则返回 RECONSUME_LATER，Mq server会在稍后重新投递这个消息，又进入步骤1。

注: Consumer 需要做好幂等控制，消息可能会被多次投递到Consumer。

0x04 其他

可以看到，极端情况下，可能仍然会出现 消费者出错的情况，不过这种情况建议人工介入处理。对于这种概率非常小的情况，使用人工介入手动处理的办法，比实现一套完整的自动回滚事务系统成本要低很多，要更划算一些。

0x05 附

1、使用限制

以下是使用Rocket mq 事务消息的一些限制:

1、事务消息不支持 delay 或者 batch操作。

2、为了避免一个 Half Message的消息被检查多次 或者 消息积压，默认对每个消息最多进行15次消息回查，可以通过修改broker的 transactionCheckMax 参数来指定次数。如果一个Half Message状态的消息检查次数超过了transactionCheckMax，默认情况下会直接丢弃掉并且打印错误日志，可以通过覆盖 AbstractTransactionCheckListener 类来修改这个行为。

3、通过 transactionMsgTimeout参数可以指定消息回查(Message Status Check)间隔。

4、事务消息可能会被check或者consume多次，要在Consumer端做好幂等控制。

2、事务状态

Rocket MQ里有三种事务状态

1、LocalTransactionState.UNKNOW : 中间状态，意味着Mq server需要稍候再次确认。

2、LocalTransactionState.COMMIT_MESSAGE: 事务完成，意味着消息可以投递给对应的 Consumer。

3、LocalTransactionState.ROLLBACK_MESSAGE: 事务失败，Mq Server会丢弃对应的事务消息，不会投递给对应的Consumer。