详述RocketMQ实现分布式事务原理+案例

快来打我* 2023-10-03 14:17 22阅读 0赞

### 一、经典的分布式事务场景 ###

示例：假设A给B转账500块，并且它俩处于不同的微服务中。  
要求：A减500块、B加上500块。

**可能发生的情况：**

1）A减500（成功），B加500（成功）
    
    2）A减500（失败），B加500（失败）
    
    3）A减500（成功），B加500（失败）
    
    4）A减500（失败），B加500（成功）

上述1）、2）两种情况是可以保证事务的一致性的。而3）、4）两种情况是无法保证事务的一致性的。而RocketMQ的分布式事务着重解决第3种情况，像第四种情况大多数时候我们可以在业务逻辑上解决。

### 二、RocketMQ实现分布式事务原理 ###

RocketMQ在4.3版本开始提供了分布式事务功能，使用RocketMQ*事务消息*来达到 ***分布式事务的最终一致性***；而不是像其他分布式事务的解决方案：2PC、3PC、TCC那样是强一致性事务。

在聊具体实现之前，我们先聊一下相关的基础概念：半消息、消息回查。

#### 1、基础概念 ####

**半消息（Half Message）：**

*  暂时无法被消费的消息。生产者将消息发送到MQ Server，但这个消息会被标记为“暂不能投递”状态，先存储起来；消费者不会去消费这条消息。
 *  只有Producer对消息进行confirm之后，消费者才会消费这条消息或丢弃这条消息。

**消息回查：**

*  网络断开或生产者重启可能会导致丢失事务消息的第二次确认。所以当MQ Server发现消息长时间处于半消息状态时，将向消息生产者发送请求，询问该消息的最终状态是提交还是回滚。

**事务消息的三种状态：**

*  Commit：提交事务消息，消费者可以消费此消息
 *  Rollback：回滚事务消息，broker会删除该消息，消费者不能消费
 *  UNKNOWN：broker需要回查确认该消息的状态。

#### 2、分布式事务交互流程 ####

如图所示：  
![请添加图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1NhaW50bW0_size_16_color_FFFFFF_t_70]  
顺着这张图，我们来理解一下整个**分布式事务的流程**：

1.  Producer发送半消息到MQ Server(Broker)，消息存储在broker里topic name为RMQ\_SYS\_TRANS\_HALF\_TOPIC的topic中。此时消息被标记为暂不能投递状态，所以消费者并不会消费。
2.  Broker存储半消息完毕之后返回ACK给Producer、表明半消息发送成功，Producer开始执行本地事务。这里执行本地事务时我们需要在业务上做一个记录，比如记录流水表；以便Broker回查确认事务消息的状态。
3.  假如Producer因为执行本地事务时宕机或者回传confirm消息过程中网络中断导致Broker没有接收到Producer的confirm消息。
4.  这时Broker会进行回查。回查的实现方式为：Broker内部开启一个定时任务，消费RMQ\_SYS\_TRANS\_HALF\_TOPIC队列中的消息，每次执行会向消息发送者确认事务执行状态（提交、回滚、未知），如果是未知则等待下一次回查。如果超过回查次数，则回滚消息。
5.  那么broker回查去哪查呢？这里我们上述所说的流水表就起到作用了，它就是用来记录Producer执行事务的状态（成功/失败）。

然后，我们思考几个问题。

**1）为什么要发送半消息？**

*  让我们知道Broker服务是否还活着，如果半消息发送失败，说明broker挂了。
 *  确保Producer端的事务执行完毕之后，consumer端事务可以执行；Producer端事务执行失败，consumer端事务则不会执行。
 *  通过回查机制保障由于网络中断、Producer宕机引起的事务消息无法确认问题。

**2）什么情况下会回查？**

*  Producer重启或宕机导致丢失事务消息的第二次确认。
 *  Producer发送二次确认消息时，网络问题导致Broker未接收到第二次确认。
 *  说白了一种是Producer没发确认消息、另一种是Producer发了确认消息但Broker没收到。

**3）为什么说RocketMQ的分布式事务是最终一致性？**

*  服务A发送确认半消息之后，服务B可以消费到消息，但是服务B可能会处理消息失败，然而服务A不关心，我已经投递消息成功了，我就认为这个事务成功了。
 *  因为consume消息有重试机制，服务B出问题只能是代码问题。如果重试消息一直都没消费成功，那么业务上应该做好补偿，保证事务的最终一致性。

原理到这里也就讲完了，这样一看RocketMQ实现分布式事务的机制还是挺简单的，后续会推出源码解析，从作者的角度去思考RocketMQ是如何设计的。

**RocketMQ通过TransactionListener的两个方法实现执行本地事务、回查机制：**

*  executeLocalTransaction --> 半消息发送成功触发此方法来执行本地事务
 *  checkLocalTransaction --> broker将发送检查消息来检查事务状态，并将调用此方法来获取本地事务状态

下面我继续讲其实现案例。

### 三、案例 ###

需求：用户在文章中心发布文章，当文章被审核通过后，用户中心给相应文章的作者增加积分。

这里不做复杂的业务逻辑介绍，仅将发送半消息、Producer执行本地事务、回查的代码片段贴出。

#### 1）maven依赖 ####

rocketmq-spring-boot-starter和RocketMQ的版本是有对应关系的，而我们的RocketMQ版本是4.8.0，所以需要作如下maven依赖处理，否者会报一些RocketMQ版本不兼容的问题。

<dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-spring-boot-starter</artifactId>
            <version>2.0.2</version>
            <exclusions>
                <exclusion>
                    <artifactId>fastjson</artifactId>
                    <groupId>com.alibaba</groupId>
                </exclusion>
                <exclusion>
                    <artifactId>commons-logging</artifactId>
                    <groupId>commons-logging</groupId>
                </exclusion>
                <exclusion>
                    <artifactId>commons-collections</artifactId>
                    <groupId>commons-collections</groupId>
                </exclusion>
                <exclusion>
                    <groupId>org.apache.rocketmq</groupId>
                    <artifactId>rocketmq-client</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>org.apache.rocketmq</groupId>
                    <artifactId>rocketmq-acl</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
    
            <dependency>
                <groupId>org.apache.rocketmq</groupId>
                <artifactId>rocketmq-client</artifactId>
                <version>4.8.0</version>
            </dependency>
            <dependency>
                <groupId>org.apache.rocketmq</groupId>
                <artifactId>rocketmq-acl</artifactId>
                <version>4.8.0</version>
            </dependency>

application.yml文件配置：

# RocketMQTemplate使用的
    rocketmq:
      name-server: 100.10.100.100:9876
      producer:
        # 小坑，必须指定group
        group: test-group

#### 2）相关基础模型 ####

1、文章审核信息类：

@Getter
    @Setter
    @NoArgsConstructor
    @AllArgsConstructor
    public class ShareAuditDTO {
        
    
        /**
         * 审核状态
         */
        private AuditStatusEnum auditStatusEnum;
    
        /**
         * 审核不通过的原因
         */
        private String reason;
    
    }

2、文章审核状态枚举类

@Getter
    @AllArgsConstructor
    public enum  AuditStatusEnum {
        
    
        /**
         * 待审核
         */
        NOT_YET,
    
        /**
         * 审核通过
         */
        PASS,
    
        /**
         * 审核不通过
         */
        REJECT;
    }

3、事务消息表

@Getter
    @Setter
    @ToString
    @Table(name = "rocketmq_transaction_log")
    @Builder
    @NoArgsConstructor
    @AllArgsConstructor
    public class RocketmqTransactionLog {
        
        /**
         * id
         */
        @Id
        @GeneratedValue(generator = "JDBC")
        private Integer id;
    
        /**
         * 事务ID
         */
        @Column(name = "transaction_Id")
        private String transactionId;
    
        /**
         * 日志
         */
        private String log;
    }

#### 3）文章中心（Producer端） ####

在文章中心服务中有一个审核文章的Controller接口，它是发送半消息的入口，如下：

@RestController
    @RequestMapping("/admin/shares")
    @Slf4j
    @RequiredArgsConstructor(onConstructor = @__(@Autowired))
    public class ShareAdminController {
        
    
        private final ShareService shareService;
    
        /**
         * 审核文章
         *
         * @param id       文章ID
         * @param auditDTO 审核信息
         * @return
         */
        @PutMapping("/audit/{id}")
        public Share auditById(@PathVariable("id") Integer id, @RequestBody ShareAuditDTO auditDTO) {
        
            return shareService.auditById(id, auditDTO);
        }
    }

1、在ShareService中是**发送半消息的逻辑**：

@Service
    @Slf4j
    @RequiredArgsConstructor(onConstructor = @__(@Autowired))
    public class ShareService {
        
    
        private final ShareMapper shareMapper;
        private final RocketMQTemplate rocketMQTemplate;
        private final RocketmqTransactionLogMapper rocketmqTransactionLogMapper;
    
        public Share auditById(Integer id, ShareAuditDTO auditDTO) {
        
            // 查询share是否存在，audit_status == not_yet
            Share share = shareMapper.selectByPrimaryKey(id);
            if (Objects.isNull(share)) {
        
                throw new IllegalArgumentException("参数非法，该分享不存在！");
            }
            if (!Objects.equals("NOT_YET", share.getAuditStatus())) {
        
                throw new IllegalArgumentException("参数非法，该分享已审核！");
            }
            // 如果是PASS，那么发送消息给RocketMQ，让用户中心去消费，并为发布人
            if (Objects.equals(AuditStatusEnum.PASS, auditDTO.getAuditStatusEnum())) {
        
    
                // 发送半消息
                String transactionId = UUID.randomUUID().toString();
    this.rocketMQTemplate.sendMessageInTransaction("tx-producer-test","add-bonus", MessageBuilder.withPayload(
                        UserAddBonusMsgDTO.builder().bonus(50)
                                .userId(share.getUserId())
                                .build())
                                // header中可以传递额外的信息
                                .setHeader(RocketMQHeaders.TRANSACTION_ID, transactionId)
                                .setHeader("share_id", id)
                                .setHeader("dto", JSONObject.toJSONString(auditDTO))
                                .build(),
                        auditDTO);
            } else {
        
                this.auditById(id, auditDTO);
            }
            return share;
        }
    
        @Transactional(rollbackFor = Exception.class)
        public void auditByIdInDB(Integer id, ShareAuditDTO auditDTO) {
        
            Share share = Share.builder().id(id)
                    .auditStatus(auditDTO.getAuditStatusEnum().toString())
                    .reason(auditDTO.getReason())
                    .build();
            shareMapper.updateByPrimaryKeySelective(share);
            // 后续可以写Redis缓存
        }
    
        /**
         * 修改数据并记录日志
         *
         * @param id            文章ID
         * @param auditDTO      文章内容
         * @param transactionId 事务ID
         */
        @Transactional(rollbackFor = Exception.class)
        public void auditByIdWithRocketMqLog(Integer id, ShareAuditDTO auditDTO, String transactionId) {
        
            auditByIdInDB(id, auditDTO);
            rocketmqTransactionLogMapper.insertSelective(RocketmqTransactionLog.builder()
                    .transactionId(transactionId)
                    .log("审核分享.....").build());
        }
    }

2、**Producer端执行本地事务以及回查**：  
类上添加@RocketMQTransactionListener注解，实现executeLocalTransaction()、checkLocalTransaction()方法，进而自定义执行本地事务逻辑和消息回查逻辑。

@RocketMQTransactionListener(txProducerGroup = "tx-producer-test")
    @RequiredArgsConstructor(onConstructor = @__(@Autowired))
    public class AddBonusTransactionListener implements RocketMQLocalTransactionListener {
        
    
        private final ShareService shareService;
        private final RocketmqTransactionLogMapper rocketmqTransactionLogMapper;
    
        /**
         * 2）执行本地事务
         */
        @Override
        public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
        
            MessageHeaders headers = message.getHeaders();
            String transactionId = (String) headers.get(RocketMQHeaders.TRANSACTION_ID);
            Integer shareId = Integer.valueOf((String) headers.get("share_id"));
            // headers中拿到的只能是字符串
            String dtoString = (String) headers.get("dto");
            ShareAuditDTO shareAuditDTO = JSON.parseObject(dtoString, ShareAuditDTO.class);
            try {
        
                shareService.auditByIdWithRocketMqLog(shareId, shareAuditDTO, transactionId);
                return RocketMQLocalTransactionState.COMMIT;
            } catch (Exception e) {
        
                return RocketMQLocalTransactionState.ROLLBACK;
            }
        }
    
        /**
         * 3）消息回查
         * 检查本地事务的状态，要有数据源去给RocketMQ检验
         */
        @Override
        public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        
            MessageHeaders headers = message.getHeaders();
            String transactionId = (String) headers.get(RocketMQHeaders.TRANSACTION_ID);
            // 查询本地记录的事务执行日志
            RocketmqTransactionLog rocketmqTransactionLog = rocketmqTransactionLogMapper.selectOne(
                    RocketmqTransactionLog.builder()
                            .transactionId(transactionId)
                            .build()
            );
            if (Objects.nonNull(rocketmqTransactionLog)) {
        
            	// 确认消息
                return RocketMQLocalTransactionState.COMMIT;
            }
            // 回滚消息
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }

#### 4）用户中心（Consumer端） ####

这里consumer没什么好说的，就是正常的消费消息，做好幂等性操作，防止消息重复消费。

@Service
    @RocketMQMessageListener(consumerGroup = "saint-consumer", topic = "add-bonus")
    @RequiredArgsConstructor(onConstructor = @__(@Autowired))
    @Slf4j
    public class AddBonusListener implements RocketMQListener<UserAddBonusMsgDTO> {
        
    
        private final UserMapper userMapper;
        private final BonusEventLogMapper bonusEventLogMapper;
    
        @Override
        public void onMessage(UserAddBonusMsgDTO message) {
        
            // 当收到消息的时候，执行的业务
            // 1. 为用户加积分
            Integer userId = message.getUserId();
            Integer bonus = message.getBonus();
            User user = userMapper.selectByPrimaryKey(userId);
            user.setBonus(user.getBonus() + bonus);
            userMapper.updateByPrimaryKey(user);
    
            // 2.记录日志到bonus_event_log表里面
            bonusEventLogMapper.insert(BonusEventLog.builder()
                    .userId(String.valueOf(userId))
                    .value(bonus)
                    .event("contribute")
                    .description("投稿。。。 加积分")
                    .createTime(new Date())
                    .build());
            log.info("积分添加完毕");
        }
    }

### 完整代码 ###

完整代码（数据库+代码）请见如下GitHub：[SpringCloudAlibabaDemo][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1NhaW50bW0_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/ca98e4a5f6834390880bb3d208946a47.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1NhaW50bW0=,size_16,color_FFFFFF,t_70
[SpringCloudAlibabaDemo]: https://github.com/Saint9768/SpringCloudAlibabaDemo