RocketMQ 实现分布式事务--可靠消息最终一致性

缺乏、安全感 2022-11-28 00:52 287阅读 0赞

**Table of Contents**

1.RocketMQ 安装

2.Rocketmq 中添加 Topic

3.可靠性消息原理

4.数据库设计

5.分布式业务搭建

1.order服务

2.account服务

3.源码

--------------------

## 1.RocketMQ 安装 ##

[安装][Link 1]

## 2.Rocketmq 中添加 Topic ##

order-topic

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjAzMjE5OQ_size_16_color_FFFFFF_t_70][]

## 3.可靠性消息原理 ##

RocketMQ 提供了可靠性消息，也叫**事务消息**。发送“可靠消息”，只需要以下三步：

1.  发送半消息（半消息不会发送给消费者）
2.  执行本地事务
3.  提交消息

完成**事务消息**发送后，消费者就可以以正常的方式来消费数据。RocketMQ 的自动重发机制在绝大多数情况下，都可以保证消息被正确消费。假如消息最终消费失败了，还可以由人工处理进行托底。

两种错误情况：

1.  事务执行失败时**回滚**消息
2.  服务器无法得知消息状态时，需要主动**回查**消息状态

## 4.数据库设计 ##

CREATE DATABASE /*!32312 IF NOT EXISTS*/`rocketmq_account` /*!40100 DEFAULT CHARACTER SET latin1 */;
    
    USE `rocketmq_account`;
    
    /*Table structure for table `account` */
    
    DROP TABLE IF EXISTS `account`;
    
    CREATE TABLE `account` (
      `id` bigint(11) NOT NULL AUTO_INCREMENT COMMENT 'id',
      `user_id` bigint(11) DEFAULT NULL COMMENT '用户id',
      `total` decimal(10,0) DEFAULT NULL COMMENT '总额度',
      `used` decimal(10,0) DEFAULT NULL COMMENT '已用余额',
      `residue` decimal(10,0) DEFAULT '0' COMMENT '剩余可用额度',
      `frozen` decimal(10,0) DEFAULT '0' COMMENT 'TCC事务锁定的金额',
      PRIMARY KEY (`id`),
      UNIQUE KEY `user_id` (`user_id`)
    ) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8;
    
    /*Data for the table `account` */
    
    insert  into `account`(`id`,`user_id`,`total`,`used`,`residue`,`frozen`) values 
    (1,1,1000,200,800,0);

CREATE DATABASE /*!32312 IF NOT EXISTS*/`rocketmq_order` /*!40100 DEFAULT CHARACTER SET latin1 */;
    
    USE `rocketmq_order`;
    
    /*Table structure for table `order` */
    
    DROP TABLE IF EXISTS `order`;
    
    CREATE TABLE `order` (
      `id` bigint(11) NOT NULL,
      `user_id` bigint(11) DEFAULT NULL COMMENT '用户id',
      `product_id` bigint(11) DEFAULT NULL COMMENT '产品id',
      `count` int(11) DEFAULT NULL COMMENT '数量',
      `money` decimal(11,0) DEFAULT NULL COMMENT '金额',
      `status` int(1) DEFAULT NULL COMMENT '订单状态：0：创建中；1：已完结',
      PRIMARY KEY (`id`)
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8;
    
    
    /*Table structure for table `tx_table` */
    
    DROP TABLE IF EXISTS `tx_table`;
    
    CREATE TABLE `tx_table` (
      `xid` char(32) NOT NULL COMMENT '事务id',
      `status` int(11) DEFAULT NULL COMMENT '0-提交，1-回滚，2-未知',
      `created_at` bigint(20) unsigned NOT NULL COMMENT '创建时间',
      PRIMARY KEY (`xid`)
    ) ENGINE=InnoDB DEFAULT CHARSET=latin1;

## 5.分布式业务搭建 ##

**通过order服务下订单，发起事务，account服务扣款**

![20200817115737873.png][]

### 1.order服务 ###

通过 create方法 发送事务消息（半消息），会触发 `TxListener` 事务监听器 执行本地事务，它执行时会回调`doCreate()` 方法完成订单的保存，本地事务执行后保存事务信息（事务id、事务状态）到数据库，以便之后进行事务回查。

事务监听器有两个方法：

*  `executeLocalTransaction()`： 执行本地事务
 *  `checkLocalTransaction()`： 负责响应Rocketmq服务器的事务回查操作

@Slf4j
    @Service
    public class OrderServiceImpl implements OrderService {
    
        @Resource
        private OrderMapper orderMapper;
        @Autowired
        private RocketMQTemplate rocketMQTemplate;
    
        @Override
        public void create(Order order) {
            // 产生事务ID
            String xid = UUID.randomUUID().toString().replace("-", "");
    
            //对事务相关数据进行封装，并转成 json 字符串
            TxAccountMessage txAccountMessage = new TxAccountMessage(order.getUserId(), order.getMoney(), xid);
            String json = JsonUtil.to(txAccountMessage);
    
            //json字符串封装到 Spring Message 对象
            Message<String> message = MessageBuilder.withPayload(json).build();
    
            //发送事务消息
            rocketMQTemplate.sendMessageInTransaction("order-topic:account",message,order);
            log.info("事务消息已发送");
        }
    
        /**
         *本地事务，执行订单保存
         *这个方法在事务监听器中调用
         * @param order
         */
        @Override
        @Transactional
        public void doCreate(Order order) {
            log.info("执行本地事务，保存订单");
            Long nextId = IdGenerator.getInstance().nextId();
            order.setId(nextId);
            orderMapper.create(order);
    
            //todo 1.模拟事务异常
    //        if (Math.random() < 0.5) {
    //            throw new RuntimeException("模拟异常");
    //        }
    
            log.info("订单已保存！ 事务日志已保存");
        }
    
    
        @Override
        public Order get(Long id) {
            return orderMapper.get(id);
        }
    }

@Slf4j
    @Component
    @RocketMQTransactionListener
    public class TxListener implements RocketMQLocalTransactionListener {
    
        @Autowired
        private OrderService orderService;
        @Resource
        private TxMapper txMapper;
    
        @Override
        public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
            log.info("事务监听 - 开始执行本地事务");
            // 监听器中得到的 message payload 是 byte[]
            String json = new String((byte[]) message.getPayload());
            String xid = JsonUtil.getString(json, "xid");
            log.info("事务监听 - " + json);
            log.info("事务监听 - xid: " + xid);
            RocketMQLocalTransactionState state;
            int status;
            Order order = (Order) o;
    
            try {
                orderService.doCreate(order);
                log.info("本地事务执行成功，提交消息");
                state = RocketMQLocalTransactionState.COMMIT;
                status = 0;
            }catch (Exception e){
                e.printStackTrace();
                log.info("本地事务执行失败，回滚消息");
                state = RocketMQLocalTransactionState.ROLLBACK;
                status = 1;
            }
            TxInfo txInfo = new TxInfo(xid, System.currentTimeMillis(), status);
            txMapper.insert(txInfo);
            return state;
        }
    
        @Override
        public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
            log.info("事务监听 - 回查事务状态");
            // 监听器中得到的 message payload 是 byte[]
            String json = new String((byte[]) message.getPayload());
            String xid = JsonUtil.getString(json, "xid");
            TxInfo txInfo = txMapper.selectById(xid);
            if (txInfo == null) {
                log.info("事务监听 - 回查事务状态 - 事务不存在："+xid);
                return RocketMQLocalTransactionState.UNKNOWN;
            }
            log.info("事务监听 - 回查事务状态 - "+ txInfo.getStatus());
            switch (txInfo.getStatus()) {
                case 0: return RocketMQLocalTransactionState.COMMIT;
                case 1: return RocketMQLocalTransactionState.ROLLBACK;
                default: return RocketMQLocalTransactionState.UNKNOWN;
            }
        }
    }

### 2.account服务 ###

`TxConsumer` 接收事务消息，调用账户业务方法 decrease，如果处理失败，Rocketmq会进行重试，直到处理成功。

@Slf4j
    @Component
    @RocketMQMessageListener(consumerGroup = "account-consumer-group",topic = "order-topic",selectorExpression = "account")
    public class TxConsumer implements RocketMQListener<String> {
    
        @Autowired
        private AccountService accountService;
    
        @Override
        public void onMessage(String message) {
            TxAccountMessage txAccountMessage = JsonUtil.from(message, new TypeReference<TxAccountMessage>() {});
            log.info("收到消息： "+txAccountMessage);
            accountService.decrease(txAccountMessage.getUserId(),txAccountMessage.getMoney());
        }
    }

### 3.源码 ###

[https://github.com/akeung/rocketmq-dtx-demo][https_github.com_akeung_rocketmq-dtx-demo]

[Link 1]: https://blog.csdn.net/weixin_42032199/article/details/108037213
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjAzMjE5OQ_size_16_color_FFFFFF_t_70]: /images/20221124/0d28732cd23043de9f8220db60f7189a.png
[20200817115737873.png]: /images/20221124/56f213e44e614a30b4221873203452cb.png
[https_github.com_akeung_rocketmq-dtx-demo]: https://github.com/akeung/rocketmq-dtx-demo