在《分布式事务：解决方案之可靠性消息最终一致性理论》中我们说到可靠消息最终一致性主要有两个解决方案：一个是本地消息表+MQ；一个是RocketMQ事务消息。今天我们重点演示一下第二个方案的实现细节。

1. 业务说明

下面我们通过消息中间件（RocketMQ）实现分布式事务来模拟两个账户的转账交易过程。交易过程是：张三给李四转账指定金额。

2. 开发环境

• 数据库：MySQL-5.6
• 微服务框架：spring-boot-2.2.2.RELEASE、spring-cloud-Hoxton.SR1、rocketmq-spring-boot-2.0.2

2.1 启动程序

源码地址：https://gitee.com/anbang713/distributed-transaction-study

启动程序之前，我们需要使用源码工程中对应的sql脚本创建相对应的数据库和表，以及插入测试数据。

（1）首先我们要启动registry-server服务注册中心；

（2）分别启动mq-demo-bank1和mq-demo-bank2微服务；

（3）启动rocketmq服务（可参考：《RocketMQ：快速入门》）。

3. 技术架构

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-qXzSumis-1598274339518)(./assets/08_技术架构.png)]

交互流程如下：

（1）Bank1向MQ Server发送转账消息。

（2）Bank1执行本地事务，扣减金额。

（3）Bank2接收转账消息，执行本地事务，添加金额。

4. 数据结构

这里除了在bank1和bank2数据库分别创建account_info表外，我们还需要再新增de_duplication交易记录表，用于交易幂等控制。

CREATE TABLE de_duplication (tx_no VARCHAR(64) NOT NULL COMMENT '事务id',create_time DATETIME,PRIMARY KEY (`tx_no`)
) ENGINE=INNODB DEFAULT CHARSET=utf8;

5. 核心代码介绍

5.1 mq-demo-bank1的核心代码

（1）向mq发送转账的事务信息

public void sendUpdateAccountBalance(AccountChangeEvent accountChangeEvent) {//将accountChangeEvent转成jsonJSONObject jsonObject =new JSONObject();jsonObject.put("accountChange",accountChangeEvent);String jsonString = jsonObject.toJSONString();//生成message类型Message<String> message = MessageBuilder.withPayload(jsonString).build();//发送一条事务消息rocketMQTemplate.sendMessageInTransaction("producer_group_txmsg_bank1","topic_txmsg", message,null);
}

（2）mq的事务监听器

@Component
@Slf4j
@RocketMQTransactionListener(txProducerGroup = "producer_group_txmsg_bank1")
public class ProducerTxMsgListener  implements RocketMQLocalTransactionListener {@Autowiredprivate AccountInfoService accountInfoService;@Autowiredprivate AccountInfoDao accountInfoDao;@Autowiredprivate DeDuplicationDao deDuplicationDao;//事务消息发送后的回调方法，当消息发送给mq成功，此方法被回调@Override@Transactionalpublic RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {try {//解析message，转成AccountChangeEventString messageString = new String((byte[]) message.getPayload());JSONObject jsonObject = JSONObject.parseObject(messageString);String accountChangeString = jsonObject.getString("accountChange");//将accountChange（json）转成AccountChangeEventAccountChangeEvent accountChangeEvent = JSONObject.parseObject(accountChangeString, AccountChangeEvent.class);//执行本地事务，扣减金额accountInfoService.doUpdateAccountBalance(accountChangeEvent);//当返回RocketMQLocalTransactionState.COMMIT，自动向mq发送commit消息，mq将消息的状态改为可消费return RocketMQLocalTransactionState.COMMIT;} catch (Exception e) {e.printStackTrace();return RocketMQLocalTransactionState.ROLLBACK;}}//事务状态回查，查询是否扣减金额@Overridepublic RocketMQLocalTransactionState checkLocalTransaction(Message message) {//解析message，转成AccountChangeEventString messageString = new String((byte[]) message.getPayload());JSONObject jsonObject = JSONObject.parseObject(messageString);String accountChangeString = jsonObject.getString("accountChange");//将accountChange（json）转成AccountChangeEventAccountChangeEvent accountChangeEvent = JSONObject.parseObject(accountChangeString, AccountChangeEvent.class);//事务idString txNo = accountChangeEvent.getTxNo();if(deDuplicationDao.existsByTxNo(txNo)){return RocketMQLocalTransactionState.COMMIT;}else{return RocketMQLocalTransactionState.UNKNOWN;}}
}

5.2 mq-demo-bank2的核心代码

（1）转账事务消息消费者

@Component
@Slf4j
@RocketMQMessageListener(consumerGroup = "consumer_group_txmsg_bank2",topic = "topic_txmsg")
public class TxMsgConsumer implements RocketMQListener<String> {@AutowiredAccountInfoService accountInfoService;//接收消息@Overridepublic void onMessage(String message) {log.info("开始消费消息:{}",message);//解析消息JSONObject jsonObject = JSONObject.parseObject(message);String accountChangeString = jsonObject.getString("accountChange");//转成AccountChangeEventAccountChangeEvent accountChangeEvent = JSONObject.parseObject(accountChangeString, AccountChangeEvent.class);//设置账号为李四的accountChangeEvent.setAccountNo("2");//更新本地账户，增加金额accountInfoService.addAccountInfoBalance(accountChangeEvent);}
}

（2）更新账户，实现转账功能

//更新账户，增加金额
@Override
@Transactional
public void addAccountInfoBalance(AccountChangeEvent accountChangeEvent) {log.info("bank2更新本地账号，账号：{},金额：{}",accountChangeEvent.getAccountNo(),accountChangeEvent.getAmount());if(deDuplicationDao.existsByTxNo(accountChangeEvent.getTxNo())){return ;}//增加金额accountInfoDao.updateAccountBalance(accountChangeEvent.getAccountNo(),accountChangeEvent.getAmount());//添加事务记录，用于幂等DeDuplication deDuplication = new DeDuplication();deDuplication.setTxNo(accountChangeEvent.getTxNo());deDuplication.setCreateTime(new Date());deDuplicationDao.save(deDuplication);if(accountChangeEvent.getAmount() == 2){throw new RuntimeException("人为制造异常");}
}

6. 测试场景

（1）张三向李四转账成功

http://localhost:9011/bank1/transfer?amount=1

（2）李四增加金额过程失败，mq将再一定时间内进行消息重复推送，消费者尝试消息实现最终一致性

http://localhost:9011/bank1/transfer?amount=2

（3）张三减少金额过程失败，本地事务回滚（此时李四还无法消费增加金额的转账消息）

http://localhost:9011/bank1/transfer?amount=3

——End—— 更多精彩分享，可扫码关注微信公众号哦。