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

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

示例：假设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、分布式事务交互流程

如图所示：

顺着这张图，我们来理解一下整个分布式事务的流程：

Producer发送半消息到MQ Server(Broker)，消息存储在broker里topic name为RMQ_SYS_TRANS_HALF_TOPIC的topic中。此时消息被标记为暂不能投递状态，所以消费者并不会消费。
Broker存储半消息完毕之后返回ACK给Producer、表明半消息发送成功，Producer开始执行本地事务。这里执行本地事务时我们需要在业务上做一个记录，比如记录流水表；以便Broker回查确认事务消息的状态。
假如Producer因为执行本地事务时宕机或者回传confirm消息过程中网络中断导致Broker没有接收到Producer的confirm消息。
这时Broker会进行回查。回查的实现方式为：Broker内部开启一个定时任务，消费RMQ_SYS_TRANS_HALF_TOPIC队列中的消息，每次执行会向消息发送者确认事务执行状态（提交、回滚、未知），如果是未知则等待下一次回查。如果超过回查次数，则回滚消息。
那么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

本文标题：详述RocketMQ实现分布式事务原理+案例

本文链接：https://blog.quwenai.cn/post/9251.html

作者授权：除特别说明外，本文由 toobug 原创编译并授权 TOOBUG信息安全网刊载发布。

本文链接：https://blog.quwenai.cn/post/9251.html

分享到：

正文

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

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

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

1、基础概念

2、分布式事务交互流程

三、案例

1）maven依赖

2）相关基础模型

3）文章中心（Producer端）

4）用户中心（Consumer端）

完整代码

相关阅读

【架构】应用架构发展历史

分布式环境下的数据一致性问题的方案讨论

【架构】分布式架构介绍及实现（简略）

【Zookeeper】应用场景：分布式锁 InterProcessMutex 源码分析

发表评论取消回复

还没有评论，来说两句吧...

目录[+]