RocketMq在阉割消息回查checkTransactionState后实现分布式事务

骑猪看日落 2021-06-10 20:40 299阅读 0赞

### **利用rocketMQ解决分布式事务** ###

**在rocketMQ中生产者有三种角色** **NormalProducer（普通）、OrderProducer（顺序）、TransactionProducer（事务）**  
**根据名字大概可以看出各个代表着什么作用，我们这里用** **TransactionProducer（事务）来解决问题。**

先举个列子来说明下我们解决方案的设计方式吧：最经典的莫过于银行转账了，网上到处都有，时序图如下  
![这里写图片描述][70]

下面贴一下测试代码：

(1) 执行业务逻辑的部分

/**
     * @Date: Created in  2018/2/12 15:55
        执行本地事务
     */
    public class TransactionExecuterimpl implements LocalTransactionExecuter{
        @Override
        public LocalTransactionState executeLocalTransactionBranch(final Message message, final Object o) {
            try{
            //DB操作 应该带上事务 service -> dao
            //如果数据操作失败  需要回滚    同事返回RocketMQ一个失败消息  意味着 消费者无法消费到这条失败的消息
            //如果成功 就要返回一个rocketMQ成功的消息，意味着消费者将读取到这条消息
            //o就是attachment
    
                //测试代码
               if(new Random().nextInt(3) == 2){
                    int a = 1 / 0;
               }
                System.out.println(new Date()+"===> 本地事务执行成功，发送确认消息");
            }catch (Exception e){
                System.out.println(new Date()+"===> 本地事务执行失败！！！");
                return LocalTransactionState.ROLLBACK_MESSAGE;
    
            }
            return LocalTransactionState.COMMIT_MESSAGE;
        }
    }

(2) 处理事务回查的代码部分

/**
     * @Date: Created in  2018/2/12 15:48
     * 未决事务，服务器端回查客户端
     */
    public class TransactionCheckListenerImpl implements TransactionCheckListener {
        @Override
        public LocalTransactionState checkLocalTransactionState(MessageExt messageExt) {
    
            System.out.println("服务器端回查事务消息： "+messageExt.toString());
            //由于RocketMQ迟迟没有收到消息的确认消息，因此主动询问这条prepare消息，是否正常？
            //可以查询数据库看这条数据是否已经处理
    
            return LocalTransactionState.COMMIT_MESSAGE;
        }
    }

(3) 启动生产者

/**
     * @Date: Created in  2018/2/12 15:24
     * 测试本地事务
     */
    public class TestTransactionProducer {
        public static void main(String[] args){
            //事务回查监听器
            TransactionCheckListenerImpl checkListener = new TransactionCheckListenerImpl();
            //事务消息生产者
            TransactionMQProducer producer = new TransactionMQProducer("transactionProducerGroup");
            //MQ服务器地址
            producer.setNamesrvAddr("192.168.56.105:9876;192.168.106:9876");
            //注册事务回查监听
            producer.setTransactionCheckListener(checkListener);
            //本地事务执行器
            TransactionExecuterimpl executerimpl = null;
            try {
                //启动生产者
                producer.start();
                executerimpl = new TransactionExecuterimpl();
                Message msg1 = new Message("TransactionTopic", "tag", "KEY1", "hello RocketMQ 1".getBytes());
                Message msg2 = new Message("TransactionTopic", "tag", "KEY2", "hello RocketMQ 2".getBytes());
    
                SendResult sendResult = producer.sendMessageInTransaction(msg1, executerimpl, null);
                System.out.println(new Date() + "msg1"+sendResult);
    
                sendResult = producer.sendMessageInTransaction(msg1, executerimpl, null);
                System.out.println(new Date() + "msg2"+sendResult);
    
            } catch (MQClientException e) {
                e.printStackTrace();
            }
            producer.shutdown();
        }
    }

(4) 消费之消费消息

/**
     * @Date: Created in  2018/2/11 15:37
     */
    public class TestConsumer {
        public static void main(String[] args) throws MQClientException {
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ConsumerGroup");
            consumer.setNamesrvAddr("192.168.56.105:9876;192.168.56.106:9876");
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            //消费普通消息
           // consumer.subscribe("TopicTest","*");
            //消费事务消息
            consumer.subscribe("TransactionTopic","*");
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                                ConsumeConcurrentlyContext context) {
                    for (MessageExt ext:msgs) {
                        try {
                            System.out.println(new Date() + new String(ext.getBody(),"UTF-8"));
                        } catch (UnsupportedEncodingException e) {
                            e.printStackTrace();
                        }
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            consumer.start();
            System.out.println("Consumer Start............");
        }
    }

**重点来了:3.2.6之前的版本这样写就可以了，但是之后的版本被关于事务回查这个借口被阉割了，不会在进行事务回查操作。**  
那么第五步向MQ发送消息如果失败的话，会造成A银行扣款成功而B银行收款未成功的数据不一致的情况

### **解决办法** ###

这里只需要考虑本地事务执行成功后的情况（因为本地事务失败不管确认消息发送成功与失败MQ集群都不会再发送消息到消费者）：

1.  本地事务成功后宕机，确认消息没有发出，分布式事务只执行一半。
2.  确认消息COMMIT\_MESSGE发出，但因网络不可达RocketMQ集群没收到。
3.  确认消息COMMIT\_MESSGE发出，RocketMQ集群收到COMMIT\_MESSGE消息，但rocketmq取消了回查机制，  
    生产者还是不知道COMMIT\_MESSGE发出是否成功。

上面三种情况的本质是一样的，就是生产者本地事务成功后，COMMIT\_MESSGE消息是否送达rocketmq集群；所以可以看做同一种情况.

下面是按key值查询事务成功提交`COMMIT_MESSGE`消息后的返回信息：

QueryResult 
    [indexLastUpdateTimestamp=1516002830440,
        messageList=[
            MessageExt [queueId=1, storeSize=291, queueOffset=0, sysFlag=4, bornTimestamp=1516002831147, bornHost=/192.168.88.1:6313, storeTimestamp=1516002830323, storeHost=/192.168.88.133:10911, msgId=C0A8588500002A9F0000000000000246, commitLogOffset=582, bodyCRC=1229495611, reconsumeTimes=0, preparedTransactionOffset=0, toString()=
                Message 
                [topic=pay, flag=0, properties=
                    {KEYS=5cf5dd03-5811-4b7f-b97c-186598e6d08b, TRAN_MSG=true, UNIQ_KEY=C0A8080B2080085EDE7B4B824F2A0000, PGROUP=transaction-pay, TAGS=tag}, 
                body=67]], 
    
            MessageExt [queueId=1, storeSize=291, queueOffset=0, sysFlag=8, bornTimestamp=1516002831147, bornHost=/192.168.88.1:6313, storeTimestamp=1516002830440, storeHost=/192.168.88.133:10911, msgId=C0A8588500002A9F0000000000000369, commitLogOffset=873, bodyCRC=1229495611, reconsumeTimes=0, preparedTransactionOffset=582, toString()=
                Message 
                [topic=pay, flag=0, properties=
                    {KEYS=5cf5dd03-5811-4b7f-b97c-186598e6d08b, TRAN_MSG=true, UNIQ_KEY=C0A8080B2080085EDE7B4B824F2A0000, PGROUP=transaction-pay, TAGS=tag},
                body=67]]
        ]
    ]

共返回两条消息：两条消息中大部分数据是一样的，但`sysFlag`、`storeTimestamp`、`msgId`、`commitLogOffset`、`preparedTransactionOffset`字段是不一样的：其中第1条为prepared发送的消息，第2条只有在提交COMMIT\_MESSGE消息成功后产生。

注意`sysFlag`、`preparedTransactionOffset`字段与`prepared`消息的区别，当提交COMMIT\_MESSGE消息成功后，推测MQ集群做了如下动作：1. 读取prepared消息，修改`sysFlag`、`preparedTransactionOffset`值，2. 在存入commitlog日志文件，设置consumerqueue序列；因为当作一条新的消息处理，所以`toreTimestamp`、`msgId`、`commitLogOffset`字段自然也就变了。所以按照发送的prepared消息的返回结果显示的`msgId`查看`sysFlag`状态只是prepared消息的`sysFlag`状态，RocketMQ4.2版本的话要用key值去查询，才能查看事务提交成功的消息标志`sysFlag=8`。  
下面是按key值查询事务失败提交`ROLLBACK_MESSAGE`消息后的返回信息：

由此我们可以根据syyFlag判断我们在提交事务以后, 消息发送是否成功

下面是逻辑分析图

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

一、在执行本地事务commit前向回查表插入消息的KEY值。

二、在生产者集群上设置一个定时任务（根据自身分布式事务流程执行的时间设定）。

1.  从回查表获取CONFIRM为0的记录列表，从记录列表中获取COUNT为3的记录，当count列达到指定阀值(假定是3)时：  
    此时记录的COUNT为3，如果CONFIRM还是为0，那么说明对此事务的回查次数为3，但RocketMQ集群还未收到COMMIT\_MESSAGE消息，说明发送COMMIT\_MESSAGE消息失败，但本地事务已经执行成功，那么必须要重发与此条记录中KEY值相对应的Perpared消息的确认消息。根据KEY值向MQ集群查找消息，根据获取的消息重新用同步的方式发送此条消息到MQ集群，并更新此记录的CONFIRM为1，COUNT+1
2.  根据第1步获取的记录列表，取出CONFIRM为0且COUNT小于3的记录，根据KEY值向MQ集群查找消息。
3.  根据第2步获取的消息判断是否是sysFlag为8的消息；如果是，更新回查表对应KEY记录的CONFIRM为1，COUNT为count+1，如果不是，更新回查表对应KEY记录的COUNT为count+1。

这里是用COUNT来避免刚刚发送消息就开始判断sysFlag的情况,相当于预留了一点rabbitmq的消息发送延迟时间,这个也可以使用CREATE\_TIME来代替,比如判断记录列表中CREATE\_TIME在当前时间前3s以前,未成功发送的消息

这只能算其中一种,还有一些手动从生产者和消费者两端进行数据库查询的方法,有兴趣的自己可以去了解

[70]: /images/20210527/6fe25ae6d54a43e09791cc565dc291f9.png
[SouthEast]: /images/20210527/bd84741b23884d99b2a7e1116e666603.png