分布式事务之可靠消息最终一致性

超、凢脫俗 2023-10-11 14:34 56阅读 0赞

### RocketMQ配置 ###

分布式事务之可靠消息最终一致性通常使用rocketmq的事务消息机制来实现。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNzUwNjU2_size_16_color_FFFFFF_t_70_pic_center]  
其大致流程如下  
1、Producer 发送事务消息  
Producer （MQ发送方）发送事务消息至MQ Server，MQ Server将消息状态标记为Prepared（预备状态），注  
意此时这条消息消费者（MQ订阅方）是无法消费到的。  
本例中，Producer 发送 ”增加积分消息“ 到MQ Server。  
2、MQ Server回应消息发送成功  
MQ Server接收到Producer 发送给的消息则回应发送成功表示MQ已接收到消息。  
3、Producer 执行本地事务  
Producer 端执行业务代码逻辑，通过本地数据库事务控制。  
本例中，Producer 执行添加用户操作。  
4、消息投递  
若Producer 本地事务执行成功则自动向MQServer发送commit消息，MQ Server接收到commit消息后将”增加积  
分消息“ 状态标记为可消费，此时MQ订阅方（积分服务）即正常消费消息；  
若Producer 本地事务执行失败则自动向MQServer发送rollback消息，MQ Server接收到rollback消息后 将删  
除”增加积分消息“ 。  
MQ订阅方（积分服务）消费消息，消费成功则向MQ回应ack，否则将重复接收消息。这里ack默认自动回应，即  
程序执行正常则自动回应ack。  
5、事务回查  
如果执行Producer端本地事务过程中，执行端挂掉，或者超时，MQ Server将会不停的询问同组的其他 Producer  
来获取事务执行状态，这个过程叫事务回查。MQ Server会根据事务回查结果来决定是否投递消息。

***启动rocketmq的nameserver***  
![在这里插入图片描述][20200909174859985.png_pic_center]  
***启动broker***

start mqbroker.cmd -n localhost:9876 autoCreateTopicEnable=true

![在这里插入图片描述][20200909174930875.png_pic_center]

### 工程实例 ###

以转账为例搭建两个工程模拟分布式事务，两个工程分别对应两个数据库  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNzUwNjU2_size_16_color_FFFFFF_t_70_pic_center 1]  
**pom依赖**

<dependency>
                <groupId>org.apache.rocketmq</groupId>
                <artifactId>rocketmq-spring-boot-starter</artifactId>
                <version>2.0.1</version>
            </dependency>

**bank1工程**

*  需要提供一个转账的controller
 *  需要提供一个listener，实现MQserver回调和回查方法
 *  转账的业务类需要提供发送消息以及保证幂等的方法

*listener监听代码*

@SuppressWarnings("all")
    @Component
    @Slf4j
    @org.apache.rocketmq.spring.annotation.RocketMQTransactionListener(txProducerGroup = MQConstans.MQ_GROUP)
    public class RocketMQTransactionListener implements RocketMQLocalTransactionListener {
        
    
        @Autowired
        private AccountInfoService accountInfoService;
    
        @Autowired
        private AccountInfoDao accountInfoDao;
    
        /**
         * 当发送事务消息成功以后回调，执行本地事务方法
         * @param message
         * @param o
         * @return
         */
        @Override
        public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
        
            if (message != null){
        
               try {
        
                   String json = new String((byte[]) message.getPayload());
                   JSONObject jsonObject = JSONObject.parseObject(json);
                   AccountEventMessage data = JSONObject.parseObject(jsonObject.getString("data"), AccountEventMessage.class);
                   log.info("回调方法开始执行,{}",data);
                   //执行本地事务
                   accountInfoService.doTransfer(data);
                   log.info("回调方法执行完成");
                   return RocketMQLocalTransactionState.COMMIT;
               }catch (Exception e){
        
                   log.info("执行回调方法出现异常,{}",e.getMessage());
                   return RocketMQLocalTransactionState.ROLLBACK;
               }
            }
            else {
        
                throw new RuntimeException("执行回调方法出现异常");
            }
        }
    
        /**
         * 事务回查方法,查看本地事务是否已经成功提交,如果提交，则会有事务号
         * @param message
         * @return
         */
        @Override
        public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        
            String json = new String((byte[]) message.getPayload());
            JSONObject jsonObject = JSONObject.parseObject(json);
            AccountEventMessage data = JSONObject.parseObject(jsonObject.getString("data"), AccountEventMessage.class);
            log.info("回查方法开始执行");
            String txNo = data.getTxNo();
            int exists = accountInfoDao.isExists(txNo);
            if (exists > 0){
        
                return RocketMQLocalTransactionState.COMMIT;
            }
            return RocketMQLocalTransactionState.UNKNOWN;
        }
    }

*转账业务类代码*

@Service
    @Slf4j
    @SuppressWarnings("ALL")
    public class AccountInfoServiceImpl implements AccountInfoService {
        
    
        @Autowired
        private AccountInfoDao accountInfoDao;
    
        @Autowired
        private RocketMQTemplate rocketMQTemplate;
    
        /**
         * 向MQ发送事务消息
         * @param accountEventMessage
         */
        @Override
        public void sendTransactionalMsg(AccountEventMessage accountEventMessage) {
        
            if (accountEventMessage != null){
        
                log.info("开始发送事务消息");
                JSONObject jsonObject = new JSONObject();
                jsonObject.put("data",accountEventMessage);
                Message<String> message = MessageBuilder.withPayload(jsonObject.toJSONString()).build();
                rocketMQTemplate.sendMessageInTransaction(MQConstans.MQ_GROUP,MQConstans.MQ_TOPIC,message,null);
                log.info("发送事务消息完成");
            }
        }
    
        @Override
        @Transactional
        public void doTransfer(AccountEventMessage accountEventMessage) {
        
            //先判断该事务号是否已经执行
            String txNo = accountEventMessage.getTxNo();
            if (StringUtils.isEmpty(txNo)){
        
                log.info("事务号不存在");
                return;
            }
            int exists = accountInfoDao.isExists(txNo);
            //幂等判断
            if (exists > 0){
        
                return;
            }
            //执行更新
            accountInfoDao.updateAccountBalance(accountEventMessage.getAccount_no(),accountEventMessage.getAmount());
            //插入事务号
            accountInfoDao.addTx(txNo);
        }
    }

*控制器代码*

@RestController
    public class UserContorller {
        
    
        @Autowired
        private AccountInfoService accountInfoService;
    
        /**
         * 执行转账
         * @param
         */
        @GetMapping("/transfer/zstols")
        public String testUserController() {
        
            String tx_no = UuidUtils.generateUuid();
            AccountEventMessage accountEventMessage = new AccountEventMessage();
            accountEventMessage.setTxNo(tx_no);
            accountEventMessage.setAccount_no("111");
            accountEventMessage.setAmount(5 * -1.0);
            accountInfoService.sendTransactionalMsg(accountEventMessage);
            return "success";
        }
    }

**bank2工程**

*  提供监听类，监听的topic要和发送端一致
 *  提供一个更新余额的方法,并在该方法中做幂等校验

监听器的代码逻辑为监听mqserver发送的消息，并调用service的方法执行更新操作

@SuppressWarnings("all")
    @Component
    @Slf4j
    //监听消息队列
    @RocketMQMessageListener(consumerGroup = "consumer_group",topic = MQConstans.MQ_TOPIC)
    public class RocketMQTransactionListenerConsumer implements RocketMQListener<String>{
        
    
    
        @Autowired
        private AccountInfoService accountInfoService;
    
        @Override
        public void onMessage(String s) {
        
            log.info("消费者开始消费消息");
            JSONObject jsonObject = JSONObject.parseObject(s);
            AccountEventMessage data = JSONObject.parseObject(jsonObject.getString("data"), AccountEventMessage.class);
            data.setAmount(5.0);
            data.setAccount_no("222");
            accountInfoService.doTransfer(data);
            log.info("消费消息完成");
        }
    }

### 总结 ###

可靠消息最终一致性就是保证消息从生产方经过消息中间件传递到消费方的一致性,RocketMQ主要解决了两个功能：  
1、本地事务与消息发送的原子性问题。  
2、事务参与方接收消息的可靠性。  
可靠消息最终一致性事务适合执行周期长且实时性要求不高的场景。引入消息机制后，同步的事务操作变为基于消  
息执行的异步操作, 避免了分布式事务中的同步阻塞操作的影响，并实现了两个服务的解耦。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNzUwNjU2_size_16_color_FFFFFF_t_70_pic_center]: https://img-blog.csdnimg.cn/20200909174657731.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNzUwNjU2,size_16,color_FFFFFF,t_70#pic_center
[20200909174859985.png_pic_center]: https://img-blog.csdnimg.cn/20200909174859985.png#pic_center
[20200909174930875.png_pic_center]: https://img-blog.csdnimg.cn/20200909174930875.png#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNzUwNjU2_size_16_color_FFFFFF_t_70_pic_center 1]: https://img-blog.csdnimg.cn/20200909175037992.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNzUwNjU2,size_16,color_FFFFFF,t_70#pic_center