在公司业务开发中，发现有的老项目使用的是低版本RocketMq，其不支持消息事务，但是下游一个比较重要的业务系统，同时又强依赖我们系统发送的mq消息(必须保证消息成功投递到Broker服务端)，因此作为上游，此种情况下,必须得保证消息一定被投递，同时又不能采用RocketMq的事务消息特性。

在此背景下，借此文章向大家介绍本地消息表的实现方案与解决思路，同时也会略带介绍RocketMq事务消息的特性

最终一致性(消息事务)

大部分的业务场景中，其实并不一定非得强求数据的强一致性，比如用户在A系统进行签到，那么在B系统里进行计算绩效，只需保证A签到成功后，B系统绩效一定会计算,至于什么时候计算,是否立马计算,其实并不重要。

其他诸如用户在A系统下了单，在B系统用户的积分一定得增加这种场景都是类似的，最终一致性也是解决分布式事务的一种常见方案。

很多场景下，我们“发消息”这个过程，目的往往是通知另外一个系统或者模块去更新数据，消息队列中的“事务”，主要解决消息生产者和消息消费者的数据一致性问题。

消息方案从本质上讲是将分布式事务转换为多个本地事务，然后依靠下游业务的重试机制达到最终一致性

错误场景一

看起来，好像消息事务就是这么点东西？上游更新完数据，发个消息让下游消费下，下游也更新下自己的数据，这不就完了？

如图所示上游肯定要保证上游数据的处理和消息的发送在一个事务里，常见的伪代码如下

@Transactional
public void execute(ActionExecuteParam param) {
	//更新本地数据库
	mapper.update(param);
	//发送mq消息通知下游更新数据
	mqProducer.sendMsg(msg);
}

上面的代码对应到执行流程图，即为如下所示，其中step3是容易被忽略的一步，如果把数据更新和发送mq消息放在一个事务里，那么实际数据的commit操作会在整个方法结束后(也就是消息完成后)进行，假如此时数据库操作出现报错、超时，那么消息已经发出去了，下游数据已经更新，但是上游数据会回滚，实际并未没有更新，出现了不一致

错误场景二

那么不加事务呢，行不行？如图所示

此时假设step1数据更新完成后，step2进行消息发送的时候，失败了，或者超时异常了，那么step1的数据不会进行回滚(数据已经commit)，此时下游数据未更新，一样出现了不一致的情况

错误场景三

先发消息，再更新数据库行不行？

这种Case就更明显了，如果消息发送成功了，但是这时候系统重启、数据库超时，都有可能导致消息发送成功(下游数据随之更新)，但是上游数据未更新，同样会出现数据不一致

所以关键点是什么？
只要上游执行的数据变更操作和发送消息不是一个原子操作，即不在一个事务中完成，那么，无论先后顺序如何，如何操作，都会出现数据不一致性问题

本地消息表

本地消息表的设计核心是将需要分布式处理的任务通过消息日志的方式来异步执行。消息发送日志(记录)可以存储到本地文本、数据库，再通过定时器自动或人工发起重试

即保证业务数据更新成功的同时，一定会有一条对应的消息记录(消息发送状态为待发送)在数据库中，然后上游所在系统单独启动一个定时器去扫描该消息表，并将状态为待发送的消息，投递到消息服务器中，失败重试，直到消息发送成功，那么就能解决数据更新和消息发送的原子性问题

整体流程图如下

这里消息发送成功时，也不一定非得说单独插一条记录到一个表里，把原始业务数据增加一个消息发送状态的字段，也是一样的效果。

如果开发中使用的消息中间件并不支持事务消息的功能，那么本地消息表是一种不错的最终一致性解决方案，那么缺点又是什么？显而易见

业务方需要单独设计消息表，及定时发送消息的定时器，增加了与业务无关的开发负担

最终一致性-RocketMq事务消息

RocketMq 4.3版本中开源了事务消息，开发者可以借此来实现简单的最终一致性。介绍事务消息之前，先抛出两个核心概念：两阶段提交、事务状态定时回查。

两阶段提交

关于两阶段提交的基本概念，贴上一张图来简单说明

因为消息发送是一个远程调用，由于网络的不稳定，无法和本地事务的执行处于一个原子操作中，针对这个缺点，RocketMQ基于两阶段提交协议做了如下改动

• 第一阶段：生产者向MQ服务器发送事务消息(prepare半消息)，服务端确认后回调通知生产者执行本地事务(此时消息为Prepare消息，存储于RMQ_SYS_TRANS_HALF_TOPIC队列中，不会被消费者消费)

• 第二阶段：生产者执行完本地事务后(业务执行完成，同时将消息唯一标记，如transactionId与该业务执行记录同时入库，方便事务回查)，根据本地事务执行结果，返回Commit/Rollback/Unknow状态码

  1、服务端若收到Commit状态码，则将prepare消息变为提交(正常消息，可被消费者消费)
  2、收到Rollback则对消息进行回滚(丢弃消息)
  3、若状态为Unknow，则等待MQ服务端定时发起消息状态回查，超过一定重试次数或者超时，消息会被丢弃

引用一张流程图来说明消息事务的两阶段提交

其中prepare半消息是事务消息的核心，正常情况下生产者投递到Broker的消息(除了延迟消息)，会立马被消费者消息，而事务消息中，需要等待生产者执行完本地事务后，才真正对半消息进行投递，这也就意味着，发送到Broker端的prepare半消息是不会被消费者立马消费到的，为什么呢？

事务消息在Broker端进行存储落盘到CommitLog的时候，会有如下2点特殊处理

1. 修改消息topic为RMQ_SYS_TRANS_HALF_TOPIC，并备份消息原有topic，供后续commit消息时还原消息topic使用
2. 修改消息queueId为0，并备份消息原有queueId，供后续commit消息时还原消息queueId使用

修改完topic和queueId后，事务消息也会像普通消息一样存储在commitLog中

看到这，是不是就明白 ，为什么prepare消息在发送后不会被立马消费？因为消息topic被修改了

事务状态定时回查

在第二阶段中，生产者在本地事务执行完成后，需要向MQ服务器返回响应状态码，发送状态码的过程也是通过Netty发送网络请求，假设由于网络原因发送失败怎么办？本地事务已经提交/回滚了，但是Commit/Rollback状态码却没发出去，那么MQ服务器上这条prepare消息状态岂不是无法被投递/回滚

因此，MQ服务端会定时扫描存储于RMQ_SYS_TRANS_HALF_TOPIC中的消息，若消息未被处理，则向消费发送者发起回调检查，检查消息对应本地事务执行状态。从而保证消息事务状态最终能和本地事务的状态一致。上图中的4、5、6就是MQ服务端定时回查步骤。

事务消息核心

RocketMq通过引入prepare半消息机制、事务消息回查机制，保证生产者消息的发送与本地事务的执行的原子性，将一个分布式大事务拆分成小事务，减少了系统间的交互。同时通过MQ 的高可用特性(不丢失)，及At-Least-Once 特性确保正确投递的事务消息会在下游一定被消费，从而保证数据的最终一致性。