史上最细最强大的RocketMQ实现分布式事务解决方案教程

本文正在参加「Java主题月 – Java 开发实战」，详情查看活动链接

前言

最近，出现了一个流行词”躺平”。

不管是大佬，还是网红，都在疯狂地讨论这个词背后的那些零零碎碎。

那么，我们在文章的开头，也说一下这个词，在我这的思考。

先说下这个词的概念，维基百科给的解释：年轻人出于对国内压抑的工作文化的失望，与其跟随社会期望坚持奋斗，不如选择“躺平”的处事态度。

不管什么时候，不管什么样的工作环境，都存在这竞争。内卷化如此严重的今天，我们该抱着怎么样的态度是生活呢？

我想躺平是一个选择，但是不免是充满了颓靡。

我想的是，你实现了你的最初的梦想了么？或许是给家人幸福，或许是买自己想买的东西，或许是见识人生中的风景。
在我看来，只要你没有实现，那么，我们就需要进一步去努力，不为别的，只为最初的梦想。人活一世，不过百年，如果不留下点什么，是不是比较遗憾。我们还是需要争取自己的未来。

或许，很多人觉得，如此内卷，怎么有未来。是的，每个人都会有这样的思考，但是，你能改变现实么？不能！你的躺平，只会在内卷的现实中，让你变成了别人的尾巴。从而，躺平，对与你来说，也做不到。

但是，又是否不要命的刺激内卷？不，我想，人要知足，但是要有小目标。这才是活的精彩的前提。

好了，回到今天的技术文章！

我们还是要进步，还是要学习知识，知识不管什么时候，都是有用的。

分布式系统架构中，少不了遇到分布式事务的实现。或许这样那样的实现，今天我们，来说一下使用消息队列中间件RocketMQ，实现分布式事务。

分布式事务定义

分布式事务就是指事务的参与者、支持事务的服务器、资源服务器以及事务管理器分别位于不同的分布式系统的不同节点之上。简单的说，就是一次大的操作由不同的小操作组成，这些小的操作分布在不同的服务器上，且属于不同的应用，分布式事务需要保证这些小操作要么全部成功，要么全部失败。本质上来说，分布式事务就是为了保证不同数据库的数据一致性。

分布式事务理论支撑

CAP

CAP定理，又被叫作布鲁尔定理。对于设计分布式系统来说(不仅仅是分布式事务)的架构师来说，CAP就是你的入门理论。以下摘自维基百科，辅助你理解C A P。

①一致性：对于客户端的每次读操作，要么读到的是最新的数据，要么读取失败。换句话说，一致性是站在分布式系统的角度，对访问本系统的客户端的一种承诺：要么我给您返回一个错误，要么我给你返回绝对一致的最新数据，不难看出，其强调的是数据正确。

②可用性：任何客户端的请求都能得到响应数据，不会出现响应错误。换句话说，可用性是站在分布式系统的角度，对访问本系统的客户的另一种承诺：我一定会给您返回数据，不会给你返回错误，但不保证数据最新，强调的是不出错。

③分区容忍性：由于分布式系统通过网络进行通信，网络是不可靠的。当任意数量的消息丢失或延迟到达时，系统仍会继续提供服务，不会挂掉。换句话说，分区容忍性是站在分布式系统的角度，对访问本系统的客户端的再一种承诺：我会一直运行，不管我的内部出现何种数据同步问题，强调的是不挂掉。

BASE

BASE 是 Basically Available(基本可用)、Soft state(软状态)和 Eventually consistent (最终一致性)三个短语的缩写。是对CAP中AP的一个扩展

基本可用:分布式系统在出现故障时，允许损失部分可用功能，保证核心功能可用。
软状态:允许系统中存在中间状态，这个状态不影响系统可用性，这里指的是CAP中的不一致。
最终一致:最终一致是指经过一段时间后，所有节点数据都将会达到一致。

BASE解决了CAP中理论没有网络延迟，在BASE中用软状态和最终一致，保证了延迟后的一致性。BASE和 ACID 是相反的，它完全不同于ACID的强一致性模型，而是通过牺牲强一致性来获得可用性，并允许数据在一段时间内是不一致的，但最终达到一致状态。

分布式事务解决方案

目前，分布式事务有很多的解决方案。相应的专门开源中间件也有，例如Seata。

今天的主人公RocketMQ事务和Seata解决的都是分布式事务问题，区别在于Seata是CAP理论，而RocketMQ方案是BASE理论也就是最终一致性。

那么，我们就完整的走一遍RocketMQ的实现分布式事务方案。

RocketMQ事务流程

执行流程如下：

Producer 即MQ发送方

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安装部署

要是使用，就需要先安装对应的消息队列服务。

下载安装

下载安装 RocketMQ 4.8.0

部署

上传安装至 /usr/local/src 目录

执行解压、安装目录指定

cd /usr/local/src
unzip rocketmq-all-4.8.0-bin-release.zip
mv rocketmq-all-4.8.0-bin-release ../rocketmq-4.8.0
复制代码

启动NameServer

cd ../rocketmq-4.8.0
nohup sh bin/mqnamesrv &
复制代码

查看启动

tail -f ~/logs/rocketmqlogs/namesrv.log
复制代码

修改Broker运行配置

vim bin/runbroker.sh

#JAVA_OPT="${JAVA_OPT} -server -Xms8g -Xmx8g -Xmn4g"
# 开发环境修改参数配置
JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m"

复制代码

启动Broker

nohup sh bin/mqbroker -n localhost:9876 &
复制代码

查看启动

tail -f ~/logs/rocketmqlogs/broker.log 
复制代码

防火墙开启端口

firewall-cmd --zone=public --add-port=9876/tcp --permanent
firewall-cmd --reload
复制代码

如果有错误，需要手动创建映射文件目录

cd  /root/store
mkdir commitlog consumequeue
复制代码

测试消息

消息发送

export NAMESRV_ADDR=localhost:9876
sh bin/tools.sh org.apache.rocketmq.example.quickstart.Producer
复制代码

消息接收

sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer
复制代码

退出运行

关闭NameServer

sh bin/mqshutdown namesrv
复制代码

关闭Broker

sh bin/mqshutdown broker
复制代码

RocketMQ控制台安装

下载地址

下载源码，完成打包

mvn clean package -Dmaven.test.skip=true
复制代码

上传至目录 /usr/local/src

脚本内容

nohup java -jar -Dspring.config.location=/app/home/rocketmq-console/application.properties /app/service/rocketmq-console/rocketmq-console-ng-2.0.0.jar >/app/home/rocketmq-console/logs/mq_console.log 2>&1 &
复制代码

开放防火墙端口

firewall-cmd --zone=public --add-port=8080/tcp --permanent
firewall-cmd --reload
复制代码

测试访问

示例业务场景

提供两个服务应用，作为消息队列的服务提供者、服务消费者

bank1 银行扣款服务

bank2 银行加款服务

场景：转账 A用户给B用户转账 A-100 B+100

bank1：

1.提供对外API

2.发起扣款请求

3.发送消息给MQ

4.MQ收到消息返回确认

5.bank1执行本地扣款业务事务并提交

MQ：

mq收到确认bank1提交后解锁消息允许消费

bank2：

1.监听MQ

2.消费消息

3.执行本地加款业务

微服务应用集成MQ

引入依赖

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>2.0.3</version>
</dependency>
复制代码

配置文件配置

rocketmq:
  name-server: xxxx:9876
  producer:
    group: base_group_syncMsg
    send-message-timeout: 5000
    retry-times-when-send-failed: 2
    max-message-size: 4194304
复制代码

bank1 应用实现

提供请求api

@GetMapping(value = "/rocketmq")
    public String transfer(@RequestParam("accountNo")String accountNo, @RequestParam("amount") Double amount){
        //创建事务id，作为消息内容发到mq
        String tx_no = UUID.randomUUID().toString();
        //封装事件实体
        AccountChangeEvent accountChangeEvent = new AccountChangeEvent(accountNo,amount,tx_no);
        //发送消息
        accountInfoService.sendUpdateAccountBalance(accountChangeEvent);
        return "处理成功-账号：{"+accountNo+"}扣减：{"+amount+"}";
    }
复制代码

扣款请求

发送消息到MQ

   /**
     * 向mq发送转账消息
     * @param accountChangeEvent 事件实体
     */
    @Override
    public void sendUpdateAccountBalance(AccountChangeEvent accountChangeEvent) {
        //将accountChangeEvent转成json
        JSONObject jsonObject =new JSONObject();
        jsonObject.put("accountChange",accountChangeEvent);
        String jsonString = jsonObject.toJSONString();
        //生成message类型
        Message<String> message = MessageBuilder.withPayload(jsonString).build();
        //发送一条事务消息
        /**
         * String txProducerGroup 生产组
         * String destination 主题，
         * Message<?> message, 消息内容
         * Object arg 参数
         */
         rocketMQTemplate.sendMessageInTransaction("producer_group_bank1","bank",message,null);
    }
复制代码

监听MQ返回

/**
 * @author 小隐乐乐
 * @date 2021/06/3
 * @description 消费者监听
 */
@Slf4j
@Component
@RocketMQMessageListener(topic = "${rocketmq.producer.topic}", consumerGroup = "${rocketmq.producer.group}")
public class ConsumerListener implements RocketMQListener<String> {

    /**
     * 注入业务实现
     */
    @Autowired
    AccountInfoService accountInfoService;

    /**
     * 接收消息
     */
    @Override
    public void onMessage(String message) {
        log.info("获取到的消费消息:{}",message);
        //解析
        JSONObject jsonObject = JSONObject.parseObject(message);
        String accountChangeString = jsonObject.getString("accountChange");
        //转成AccountChangeEvent对象
        AccountChangeEvent accountChangeEvent = JSONObject.parseObject(accountChangeString, AccountChangeEvent.class);
        //设置账号
        accountChangeEvent.setAccountNo("2");
        //执行业务操作---增加金额
        accountInfoService.addAccountInfoBalance(accountChangeEvent);
    }
}
复制代码

实现本地业务逻辑

/**
 * @author 小隐乐乐
 * @date 2021/06/3
 * @description 账户业务实现
 */
@Service
@Slf4j
public class AccountInfoServiceImpl implements AccountInfoService {

    @Autowired
    AccountInfoDao accountInfoDao;

    //更新账户--增加金额
    @Override
    @Transactional
    public void addAccountInfoBalance(AccountChangeEvent accountChangeEvent) {
        log.info("bank2更新本地账号，账号：{},金额：{}",accountChangeEvent.getAccountNo(),accountChangeEvent.getAmount());
        //本地读取事务  防止重复消费
        if(accountInfoDao.isExistTx(accountChangeEvent.getTxNo())>0){
            return ;
        }
        //插入数据--增加金额
        accountInfoDao.updateAccountBalance(accountChangeEvent.getAccountNo(),accountChangeEvent.getAmount());
        //添加事务记录，用于幂等
        accountInfoDao.addTx(accountChangeEvent.getTxNo());
        //预留错误演示
        if(accountChangeEvent.getAmount() == 250){
            throw new RuntimeException("消息处理异常");
        }
    }
}
复制代码

bank1 事务回调监听

/**
 * @author 小隐乐乐
 * @date 2021/06/3
 * @description 生产者事务回调监听器
 */
@Component
@Slf4j
@RocketMQTransactionListener(txProducerGroup = "producer_group_bank1")
public class ProducerCallbackListener implements RocketMQLocalTransactionListener {

    @Autowired
    AccountInfoService accountInfoService;

    @Autowired
    AccountInfoDao accountInfoDao;

    /**
     * 事务消息发送后的回调方法，当消息发送给mq成功，此方法被回调
     * @param message 消息
     * @return
     */
    @Override
    @Transactional
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
        try {
            //解析消息
            String messageString = new String((byte[]) message.getPayload());
            JSONObject jsonObject = JSONObject.parseObject(messageString);
            //转成AccountChangeEvent实体
            String accountChangeString = jsonObject.getString("accountChange");
            //将accountChange（json）转成AccountChangeEvent
            AccountChangeEvent accountChangeEvent = JSONObject.parseObject(accountChangeString, AccountChangeEvent.class);
            //执行本地事务，扣减金额
            accountInfoService.doUpdateAccountBalance(accountChangeEvent);
            //当返回RocketMQLocalTransactionState.COMMIT，自动向mq发送commit消息，mq将消息的状态改为可消费
            return RocketMQLocalTransactionState.COMMIT;
        } catch (Exception e) {
            e.printStackTrace();
            //向mq发送ROLLBACK，mq将消息的状态依旧无法消费
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }

    /**
     * 事务状态回查，查询是否扣减金额
     * @param message 消息
     * @return
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        //解析message，转成AccountChangeEvent
        String messageString = new String((byte[]) message.getPayload());
        JSONObject jsonObject = JSONObject.parseObject(messageString);
        String accountChangeString = jsonObject.getString("accountChange");
        //将accountChange（json）转成AccountChangeEvent
        AccountChangeEvent accountChangeEvent = JSONObject.parseObject(accountChangeString, AccountChangeEvent.class);
        //事务id
        String txNo = accountChangeEvent.getTxNo();
        int existTx = accountInfoDao.isExistTx(txNo);
        if(existTx>0){
            return RocketMQLocalTransactionState.COMMIT;
        }else{
            return RocketMQLocalTransactionState.UNKNOWN;
        }
    }
}
复制代码

bank2 应用实现

bank2 监听MQ

/**
 * @author 小隐乐乐
 * @date 2021/06/3
 * @description 消费者监听
 */
@Slf4j
@Component
@RocketMQMessageListener(topic = "bank", consumerGroup = "rocketmq.consumer.group")
public class ConsumerListener implements RocketMQListener<String> {

    /**
     * 注入业务实现
     */
    @Autowired
    AccountInfoService accountInfoService;

    /**
     * 接收消息
     */
    @Override
    public void onMessage(String message) {
        log.info("获取到的消费消息:{}",message);
        //解析
        JSONObject jsonObject = JSONObject.parseObject(message);
        String accountChangeString = jsonObject.getString("accountChange");
        //转成AccountChangeEvent对象
        AccountChangeEvent accountChangeEvent = JSONObject.parseObject(accountChangeString, AccountChangeEvent.class);
        //设置账号
        accountChangeEvent.setAccountNo("2");
        //执行业务操作---增加金额
        accountInfoService.addAccountInfoBalance(accountChangeEvent);
    }
}
复制代码

消息消费

log.info("获取到的消费消息:{}",message);
        //解析
        JSONObject jsonObject = JSONObject.parseObject(message);
        String accountChangeString = jsonObject.getString("accountChange");
        //转成AccountChangeEvent对象
        AccountChangeEvent accountChangeEvent = JSONObject.parseObject(accountChangeString, AccountChangeEvent.class);
复制代码

执行本地扣款事务

accountInfoService.addAccountInfoBalance(accountChangeEvent);
复制代码

总结

终于搞完了，写demo还是很费事的。

分布式事务解决方案很多，到底需不需要分布式事务，也是需要我们技术人员去考量的。那么如果需要，我相信，本篇文章作为RocketMQ实现消息队列分布式事务的快速上手文章，相信你不容错过。如果觉得写的不错，我准备出专栏，哈哈哈。

躺平，在追求梦想的人身上不是一个好选择，技术的脚步，是一直向前的，努力吧，少年们！！！

文章版权归作者所有，未经允许请勿转载。

THE END

后端

Symantec Gateway Security未明远程DNS缓存中毒漏洞

Flutter桌面端开发——复制和粘贴内容

从源码角度来研究PostgreSQL是如何管理使用文件描述符

PyMongo 修改数据

前端面试必不可少的基础知识，虽然少但是你不能不知道

LBlog ‘Comments.ASP’ SQL注入漏洞

史上最细最强大的RocketMQ实现分布式事务解决方案教程｜Java 开发实战

前言