分布式事务解决⽅案之可靠消息最终⼀致性
5.1.什么是可靠消息最终⼀致性事务
  可靠消息最终⼀致性⽅案是指当事务发起⽅执⾏完成本地事务后并发出⼀条消息，事务参与⽅(消息消费者)⼀定能够接收消息并处理事务成功，此⽅案强调的是只要消息发给事务参与⽅最终事务要达到⼀致。
此⽅案是利⽤消息中间件完成，如下图：
  事务发起⽅（消息⽣产⽅）将消息发给消息中间件，事务参与⽅从消息中间件接收消息，事务发起⽅和消息中间件之间，事务参与⽅（消息消费⽅）和消息中间件之间都是通过⽹络通信，由于⽹络通信的不确定性会导致分布式事务问题。
因此可靠消息最终⼀致性⽅案要解决以下⼏个问题：
1.本地事务与消息发送的原⼦性问题
  本地事务与消息发送的原⼦性问题即：事务发起⽅在本地事务执⾏成功后消息必须发出去，否则就丢弃消息。即实现本地事务和消息发送的原⼦性，要么都成功，要么都失败。本地事务与消息发送的原⼦性问题是实现可靠消息最终⼀致性⽅案的关键问题。
先来尝试下这种操作，先发送消息，再操作数据库：
begin transaction；
//1.发送MQ
//2.数据库操作
commit transation;
  这种情况下⽆法保证数据库操作与发送消息的⼀致性，因为可能发送消息成功，数据库操作失败。
你⽴马想到第⼆种⽅案，先进⾏数据库操作，再发送消息：
begin transaction；
//1.数据库操作
//2.发送MQ
commit transation;
这种情况下貌似没有问题，如果发送MQ消息失败，就会抛出异常，导致数据库事务回滚。但如果是超时异常，数据库回滚，但MQ其实已经正常发送了，同样会导致不⼀致。
2、事务参与⽅接收消息的可靠性
  事务参与⽅必须能够从消息队列接收到消息，如果接收消息失败可以重复接收消息。
3、消息重复消费的问题
  由于⽹络2的存在，若某⼀个消费节点超时但是消费成功，此时消息中间件会重复投递此消息，就导致了消息的重复消费。
要解决消息重复消费的问题就要实现事务参与⽅的⽅法幂等性。
5.2.解决⽅案
  上节讨论了可靠消息最终⼀致性事务⽅案需要解决的问题，本节讨论具体的解决⽅案。
5.2.1.本地消息表⽅案
  本地消息表这个⽅案最初是eBay提出的，此⽅案的核⼼是通过本地事务保证数据业务操作和消息的⼀致性，然后通过定时任务将消息发送⾄消息中间件，待确认消息发送给消费⽅成功再将消息删除。
下⾯以注册送积分为例来说明：
下例共有两个微服务交互，⽤户服务和积分服务，⽤户服务负责添加⽤户，积分服务负责增加积分。
交互流程如下：
  1、⽤户注册
  ⽤户服务在本地事务新增⽤户和增加 ”积分消息⽇志“。（⽤户表和消息表通过本地事务保证⼀致）
下边是伪代码
begin transaction；
//1.新增⽤户
//2.存储积分消息⽇志
commit transation;
这种情况下，本地数据库操作与存储积分消息⽇志处于同⼀个事务中，本地数据库操作与记录消息⽇志操作具备原⼦性。
  2、定时任务扫描⽇志
  如何保证将消息发送给消息队列呢？
经过第⼀步消息已经写到消息⽇志表中，可以启动独⽴的线程，定时对消息⽇志表中的消息进⾏扫描并发送⾄消息中间件，在消息中间件反馈发送成功后删除该消息⽇志，否则等待定时任务下⼀周期重试。
  3、消费消息
  如何保证消费者⼀定能消费到消息呢？
  这⾥可以使⽤MQ的ack（即消息确认）机制，消费者监听MQ，如果消费者接收到消息并且业务处理完成后向MQ发送ack（即消息确认），此时说明消费者正常消费消息完成，MQ将不再向消费者推送消息，否则消费者会不断重试向消费者来发送消息。
  积分服务接收到”增加积分“消息，开始增加积分，积分增加成功后向消息中间件回应ack，否则消息中间件将重复投递此消息。
由于消息会重复投递，积分服务的”增加积分“功能需要实现幂等性。
5.2.2.RocketMQ事务消息⽅案
  RocketMQ 是⼀个来⾃阿⾥巴巴的分布式消息中间件，于 2012 年开源，并在 2017 年正式成为 Apache 顶级项⽬。据了解，包括阿⾥云上的消息产品以及收购的⼦公司在内，阿⾥集团的消息产品全
线都运⾏在 RocketMQ 之上，并且最近⼏年的双⼗⼀⼤促中，RocketMQ 都有抢眼表现。Apache RocketMQ 4.3之后的版本正式⽀持事务消息，为分布式事务实现提供了便利性⽀持。
  RocketMQ 事务消息设计则主要是为了解决 Producer 端的消息发送与本地事务执⾏的原⼦性问题，RocketMQ 的设计中 broker 与producer 端的双向通信能⼒，使得 broker 天⽣可以作为⼀个事务协调者存在；⽽ RocketMQ 本⾝提供的存储机制为事务消息提供了持久化能⼒；RocketMQ 的⾼可⽤机制以及可靠消息设计则为事务消息在系统发⽣异常时依然能够保证达成事务的最终⼀致性。
  在RocketMQ 4.3后实现了完整的事务消息，实际上其实是对本地消息表的⼀个封装，将本地消息表移动到了MQ内部，解决 Producer 端的消息发送与本地事务执⾏的原⼦性问题。
springboot其实就是spring
执⾏流程如下：
  为⽅便理解我们还以注册送积分的例⼦来描述整个流程。
Producer 即MQ发送⽅，本例中是⽤户服务，负责新增⽤户。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实现，对⽤户侧来说，⽤户需要分别实现本地事务执⾏以及本地事务回查⽅法，因此只需关注本地事务的执⾏状态即可。
RoacketMQ提供RocketMQLocalTransactionListener接⼝：
public interface RocketMQLocalTransactionListener {
/**
‐发送prepare消息成功此⽅法被回调，该⽅法⽤于执⾏本地事务
‐    @param msg 回传的消息，利⽤transactionId即可获取到该消息的唯⼀Id
‐    @param arg 调⽤send⽅法时传递的参数，当send时候若有额外的参数可以传递到send⽅法中，这⾥能获取到
‐    @return 返回事务状态，COMMIT：提交 ROLLBACK：回滚 UNKNOW：回调 */
RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg);
/**
‐    @param msg 通过获取transactionId来判断这条消息的本地事务执⾏状态
‐    @return 返回事务状态，COMMIT：提交 ROLLBACK：回滚 UNKNOW：回调 */
RocketMQLocalTransactionState checkLocalTransaction(Message msg);
}
发送事务消息：
以下是RocketMQ提供⽤于发送事务消息的API：
TransactionMQProducer producer = new TransactionMQProducer("ProducerGroup"); producer.setNamesrvAddr("127.0.0.1:9876"); producer.start();
//设置TransactionListener实现
producer.setTransactionListener(transactionListener）；
//发送事务消息
SendResult sendResult = producer.sendMessageInTransaction(msg, null);
5.3.RocketMQ实现可靠消息最终⼀致性事务
5.3.1.业务说明
  本实例通过RocketMQ中间件实现可靠消息最终⼀致性分布式事务，模拟两个账户的转账交易过程。
两个账户在分别在不同的银⾏(张三在bank1、李四在bank2)，bank1、bank2是两个微服务。交易过程是，张三给李四转账指定⾦额。
上述交易步骤，张三扣减⾦额与给bank2发转账消息，两个操作必须是⼀个整体性的事务。
5.3.2.程序组成部分
本⽰例程序组成部分如下：
  数据库：MySQL-5.7.25，包括bank1和bank2两个数据库。
  JDK：64位 jdk1.8.0_201
  rocketmq 服务端：RocketMQ-4.5.0
  rocketmq 客户端：RocketMQ-Spring-Boot-starter.2.0.2-RELEASE
  微服务框架：spring-boot-2.1.3、spring-cloud-Greenwich.RELEASE
  微服务及数据库的关系：
    dtx/dtx-txmsg-demo/dtx-txmsg-demo-bank1 银⾏1，操作张三账户，连接数据库bank1
    dtx/dtx-txmsg-demo/dtx-txmsg-demo-bank2 银⾏2，操作李四账户，连接数据库bank2
本⽰例程序技术架构如下：
交互流程如下：
  1、Bank1向MQ Server发送转账消息
  2、Bank1执⾏本地事务，扣减⾦额
  3、Bank2接收消息，执⾏本地事务，添加⾦额
5.3.3.创建数据库
  导⼊数据库脚本：资料\sql\bank1.sql、资料\sql\bank2.sql，已经导过不⽤重复导⼊。
创建bank1库，并导⼊以下表结构和数据(包含张三账户)
CREATE DATABASE `bank1` CHARACTER SET 'utf8' COLLATE 'utf8_general_ci';
DROP TABLE IF EXISTS `account_info`;
CREATE TABLE `account_info`    (
`id` bigint(20) NOT NULL AUTO_INCREMENT,
`account_name` varchar(100) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL COMMENT '户主姓名',
`account_no` varchar(100) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL COMMENT '银⾏卡号',
`account_password` varchar(100) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL
COMMENT '帐户密码',    `account_balance` double NULL DEFAULT NULL COMMENT '帐户余额', PRIMARY KEY (`id`) USING BTREE
)    ENGINE = InnoDB AUTO_INCREMENT = 5 CHARACTER SET = utf8 COLLATE = utf8_bin ROW_FORMAT = Dynamic; INSERT INTO `account_info` VALUES (2, '张三的账户', '1', '', 10000);
创建bank2库，并导⼊以下表结构和数据(包含李四账户)
CREATE DATABASE `bank2` CHARACTER SET 'utf8' COLLATE 'utf8_general_ci';
CREATE TABLE `account_info`    (
`id` bigint(20) NOT NULL AUTO_INCREMENT,
`account_name` varchar(100) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL COMMENT '户主姓名',
`account_no` varchar(100) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL COMMENT '银⾏卡号',
`account_password` varchar(100) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL COMMENT '帐户密码',    `account_balance` double NULL DEFAULT NULL COMMENT '帐户余额', PRIMARY KEY (`id`) USING BTREE
)    ENGINE = InnoDB AUTO_INCREMENT = 5 CHARACTER SET = utf8 COLLATE = utf8_bin ROW_FORMAT = Dynamic;
INSERT INTO `account_info` VALUES (3, '李四的账户', '2', NULL, 0);
在bank1、bank2数据库中新增de_duplication，交易记录表(去重表)，⽤于交易幂等控制。DROP TABLE IF EXISTS `de_duplication`;
CREATE TABLE `de_duplication`    (
`tx_no`    varchar(64) COLLATE utf8_bin NOT NULL,
`create_time` datetime(0) NULL DEFAULT NULL,
PRIMARY KEY (`tx_no`) USING BTREE
) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_bin ROW_FORMAT = Dynamic;
5.3.4.启动RocketMQ
（1）下载RocketMQ服务器
（2）解压并启动
启动nameserver:
set ROCKETMQ_HOME=[rocketmq服务端解压路径]
start [rocketmq服务端解压路径]/d
启动broker:
set ROCKETMQ_HOME=[rocketmq服务端解压路径]
start [rocketmq服务端解压路径]/d ‐n 127.0.0.1:9876 autoCreateTopicEnable=true
3.3.5 导⼊dtx-txmsg-demo
  dtx-txmsg-demo是本⽅案的测试⼯程，根据业务需求需要创建两个dtx-txmsg-demo⼯程。（1）导⼊dtx-txmsg-demo
导⼊：资料\基础代码\dtx-txmsg-demo到⽗⼯程dtx下。
两个测试⼯程如下：
  dtx/dtx-txmsg-demo/dtx-txmsg-demo-bank1 ，操作张三账户，连接数据库bank1
  dtx/dtx-txmsg-demo/dtx-txmsg-demo-bank2 ，操作李四账户，连接数据库bank2
（2）⽗⼯程maven依赖说明
在dtx⽗⼯程中指定了SpringBoot和SpringCloud版本
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring‐boot‐dependencies</artifactId>
<version>2.1.3.RELEASE</version>
<type>pom</type>
<scope>import</scope>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring‐cloud‐dependencies</artifactId>
<version>Greenwich.RELEASE</version>
<type>pom</type>
<scope>import</scope>
</dependency>
在dtx-txmsg-demo⽗⼯程中指定了rocketmq-spring-boot-starter的版本。
<dependency>
<groupId>ketmq</groupId>
<artifactId>rocketmq‐spring‐boot‐starter</artifactId>
<version>2.0.2</version>
</dependency>
（3）配置rocketMQ
在application-local.propertis中配置rocketMQ nameServer地址及⽣产组：
up = producer_bank2
rocketmq.name‐server = 127.0.0.1:9876

版权声明：本站内容均来自互联网，仅供演示用，请勿用于商业和其他非法用途。如果侵犯了您的权益请与我们联系QQ:729038198，我们将在24小时内删除。