RocketMQ事务消息回查设计方案

矫情吗；* 2022-05-28 10:18 774阅读 0赞

先看阿里云消息队列的流程示意图：

![阿里云消息队列][SouthEast]

红色的⑤，⑥，⑦步骤就是3.1.2之前的有事务消息回查功能的版本。消息回查功能基于文件系统，回查后得到的结果以及正常的处理结果Commit/Rollback都会修改CommitLog里PREPARED消息的状态，这会导致内存中脏页过多，有隐患。在之后的版本移除了基于文件系统的状态修改机制，对事务消息的处理流程进行重做，移除了消息回查功能。

阿里云的事务消息回查功能基于数据库机制，不过目前未开源。

目前版本的事务消息流程如下：

1.  先发送PREPARED消息，返回其CommitLog Offset
2.  执行本地逻辑，得到处理结果是Commit还是Rollback
3.  将处理结果和CommitLog Offset发送到Broker
4.  Broker先根据Offset从CommitLog中提起PREPARED消息，然后克隆此消息生成新的消息，消息Body(内容)和PREPARED的一致。先设置处理结果标识，然后根据处理结果，如果是Rollback，则清空body，否则不清空，最后存储进CommitLog
5.  Broker在存储PREPARED消息时，不会将其PositionInfo存入ConsumeQueue，也就是正常情况下此消息不会被消费；但会为其生成IndexInfo存入IndexFile，也就是能通过key查询此消息。
6.  当事务处理结果是Rollback时，克隆消息不会生成PositionInfo和IndexInfo，所以此消息不会被消费，不能被查询，事务流程就此结束；当事务处理结果是Commit时，克隆消息会生成PositionInfo和IndexInfo，也就是能被正常消费，也能正常查询。
    
    事务消息的确认(Commit/Rollback)的执行方式是Oneway形式，也就是单向执行，没有结果返回，这种形式执行效率很高，但是有个问题就是不确定确认操作是否执行成功，可能因为网络问题或者Broker问题造成发送失败，消息回查就是解决这个问题，但现在没有了，所以需要我们自己设计解决。

事务确认操作失败，有两种补偿操作，**一种是重新发送一次事务消息，另一种是去查询PREPARED消息**。

相对于第一种方案，第二种在操作上相对要简单些，其实现需要依赖RocketMQ的事务执行机制。当执行Rollback后，事务消息作废，不会被消费到；而Commit的话，其存储的MessageBody和PREPARED是发送的一致，也就是说消息的uniqueKey一致，这个属性在发送时由发送者Client生成，存入Message中，每个消息都有。Broker会将此Index信息存入IndexFile，之后客户端可以通过此key去查询相应的消息。

发送事务消息时，无论成功与否，都会返回PREPARED的uniqueKey：

/** * 发送事务消息 * * @param msg 消息 * @param tranExecuter 【本地事务】执行器 * @param arg 【本地事务】执行器参数 * @return 事务发送结果 * @throws MQClientException 当 Client 发生异常时 */
    public TransactionSendResult sendMessageInTransaction(......){
        ......
        // 返回【事务发送结果】
        TransactionSendResult transactionSendResult = new TransactionSendResult();
        transactionSendResult.setSendStatus(sendResult.getSendStatus());
        transactionSendResult.setMessageQueue(sendResult.getMessageQueue());
        //提取Prepared消息的uniqID
        transactionSendResult.setMsgId(sendResult.getMsgId());
        transactionSendResult.setQueueOffset(sendResult.getQueueOffset());
        transactionSendResult.setTransactionId(sendResult.getTransactionId());
        transactionSendResult.setLocalTransactionState(localTransactionState);
        return transactionSendResult;
    
    }

**Commit消息和Prepared消息的uniqID(uniqueKey)一致，且能通过uniqID去查询Prepared消息是如下设计实现的前提！**

![事务回查设计][SouthEast 1]

下面按照设计图上的流程逐步的说明设计思路：

1.  Producer在处理本地事务逻辑是，如果返回的结果是Commit，在向Broker提交Commit操作后，提取SendRetuls里的msgId，其值是Prepared消息的uniqueKey。在Broker创建好一个topic，专门用于传输uniqueKey，Producer向Broker发送uniqueKey(非事务形式)。
2.  Consumer订阅事务消息topic和uniqueKey的topic。
3.  将消费事务消息和存储此消息的msgId进缓存(Redis)中间件作为一个原子操作，消息消费成功，则事务消息的msgId存入Redis的集合中。
4.  同理，在消费了Prepared的uniqueKey时，将其值存入Redis的另一个集合中。
5.  Commit Message和Half Message的MessageBody是相同的，因此其uniqueKey也相同，都是发送Half Message时Client生成的，也就是说Redis里两个Set的内容大部分重合。但事务消息的确认操作可能会失败，此举会造成Commit msgId丢失，所以Set< Half msgId> 包含 Set< Commit msgId>。
6.  定时计算 Set< Half msgId> 与 Set< Commit msgId>的差值集合，在计算完的同时删除其key，可以使用Redis的multi事务来保证。
7.  将得到的差值集合，也就是msgId集合，通过RocketMQ Client的viewMessage(topic，msgId) 方法，查询Prepared消息。
8.  但是消费事务消息和消费uniqueKey消息的顺序不能确定，可能出现如下情况：

![这里写图片描述][SouthEast 2]

差值集合的元素不一定就是事务确认操作失败而丢失的msgId，也可能是定时任务执行时元素错位而引起的，所以通过差值集合获取到的Prepared消息时，实际丢失的消息不会遗漏，但可能会有因为元素错位而增加的，执行时需要根据msgId在业务上去重。

定时任务可以仅由一台服务器执行，也可以多台服务器并行，对msgId的的最终结果应该没有影响。

当消费者集群中的某台服务器宕机时，消费队列负载均衡，事务消息或者uniqueKey消息可能有重复，但不会丢失，其他消费者正常消费。

当Redis不可用时，挂起uniqueKey的消费。重启时，扫描宕机后数据库新插入的msgId，存入Redis作为  
Set< Commit msgId>元素，然后重启uniqueKey的消费，生成Set< Half msgId>，差值计算后去重后消费。

[SouthEast]: /images/20220528/450a30973e644c3983645d581aed93dd.png
[SouthEast 1]: /images/20220528/c0e4db571de5490ca3a3877b938dcd2f.png
[SouthEast 2]: /images/20220528/db7135123b19473580dc4dabdfa00166.png