分布式事务之解决方案（可靠消息最终一致性）

时间  2019-11-29
标签 分布式 事务 解决 方案 可靠消息 最终 一致性 栏目 系统架构 繁體版
原文   原文链接

 

5. 分布式事务解决方案之可靠消息最终一致性

5.1. 什么是可靠消息最终一致性事务

可靠消息最终一致性方案是指当事务发起执行彻底本地事务后并发出一条消息，事务参与方（消息消费者）必定可以接收消息并处理事务成功，此方案强调的是只要消息发给事务参与方最终事务要达到一致。
此方案是利用消息中间件完成，以下图：
事务发起方（消息生产方）将消息发给消息中间件，事务参与方从消息中间件接收消息，事务发起方和消息中间件之间，事务参与方（消息消费方）和消息中间件之间都是经过网络通讯，因为网络通讯的不肯定性致使分布式事务问题。
在这里插入图片描述
所以可靠消息最终一致性方案要解决如下几个问题 ：
一、本地事务与消息发送的原子性问题
本地事务与消息发送的原子性问题即 ：事务发起方在本地事务执行成功后消息必须发出去，不然就丢弃消息。即实现本地事务和消息发送的原子性，要么都成功，要么都失败。本地事务与消息发送的原子性问题是实现可靠消息最终一致性方案的关键问题。
先来尝试下这种操做，先发送消息，再操做数据库 ：java

begin transaction；
		// 1.发送MQ
		// 2.数据库操做
commit transation；

这种状况下没法保证数据库操做与发送消息的一致性，由于可能发送消息成功，数据库操做失败。
你立马想到第二种方案，先进行数据库操做，再发送消息 ：spring

begin transaction；
		// 1.数据库操做
		// 2.发送MQ
commit transation；

这种状况下貌似没有问题，若是发送MQ消息失败，就会抛出异常，致使数据库事务回滚。但若是是超时异常，数据库回滚，但MQ其实已经正常发送来，一样会致使不一致。
二、事务参与方接收消息的可靠性
事务参与方必须可以从消息队列接收到消息，若是接收消息失败能够重复接收消息。
三、消息重复消费的问题
因为网络2的存在，若某一个消费节点超时可是消费成功，此时消息中间件会重复投递此消息，就致使来消息的重复消费。要解决消息重复消费的问题就要实现事务参与方的方法幂等性。数据库

5.2. 解决方案

5.2.1. 本地消息表方案

本地消息表这个方案最初是eBay提出的，此方案的核心是经过本地事务保证数据业务操做和消息的一致性，而后经过定时任务将消息发送至消息中间件，待确认消息发送给消费方成功再将消息删除。
下面以注册送积分为例来讲明 ：
下例共有两个微服务交互，用户服务和积分服务，用户服务负责添加用户，积分服务负责增长积分。
在这里插入图片描述
交互流程以下 ：
一、用户注册
用户服务在本地事务新增用户和增长“积分消息日志”。（用户表和消息表经过本地事务保证一致）
下表是伪代码apache

begin transaction；
		// 1.新增用户
		// 2.存储积分消息日志
commit transation；

这种状况下，本地数据库操做与存储积分消息日志处于同一事务中，本地数据库操做与记录消息日志操做具有原子性。
二、定时任务扫描日志
如何保证将消息发送给消息队列呢？
通过第一步消息已经写到消息日志表中，能够启动独立的线程，定时对消息日志表中的消息进行扫描并发送至消息中间件，在消息中间件反馈发送成功后删除该消息日志，不然等待定时任务下一周期重试。
三、消费消息
如何保证消费者必定能消费到消息呢？
这里可使用MQ的ack（即消息确认）机制，消费者监听MQ，若是消费者接收到消息而且业务处理完成后向MQ发送ack（即消息确认），此时说明消费者正常消费消息完成，MQ将再也不向消费者推送消息，不然消费者会不断重试向消费者来发送消息。
积分服务接收到“增长积分”消息，开始增长积分，积分增长成功后消息中间件回应ack，不然消息中间件将重复投递此消息。
因为消息会重复投递，积分服务的“增长积分”功能须要实现幂等性。json

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端的消息发送与本地事务执行的原子性问题。
在这里插入图片描述
执行流程以下 ：
为方便理解咱们还以注册送积分的例子来描述整个流程。
Producer即MQ发送方，本例中是用户服务，负责新增用户。MQ订阅方即消息消费方，本例中是积分服务，负责新增积分。
一、Producer发送事务消息
Producer（MQ发送方）发送事务消息至MQ Server，MQ Server将消息状态标记为Prepared（预览状态），注意此时这条消息消费者（MQ订阅方）是没法消费到的。
二、MQ Server回应消息发送成功
MQ Server接收到Producer发送给的消息则回应发送成功表示MQ已接收到消息。
三、Producer执行本地事务
Producer端执行业务代码逻辑，经过本地数据库事务控制。
本例中，Producer执行添加用户操做。
四、消息投递
若Producer本地事务执行成功则自动向MQ Server发送commit消息，MQ Server接收到commit消息后将“增长积分消息”状态标记为可消费，此时MQ订阅方（积分服务）即正常消费消息；
若Producer 本地事务执行失败则自动向MQ Server发送rollback消息，MQ Server接收到rollback消息后将删除“增长积分消息”。
MQ订阅方（积分服务）消费消息，消费成功则向MQ回应ack，不然将重复接收消息。这里ack默认自动回应，即程序执行正常则自动回应ack。
五、事务回查
若是执行Producer端本地事务过程当中，执行端挂掉，或者超时，MQ Server将会不停的询问同组的其余Producer来获取事务执行状态，这个过程叫事务回查。MQ Server会根据事务回查结果来决定是否投递消息。
以上主干流程已由RocketMQ实现，对用户则来讲，用户须要分别实现本地事务执行以及本地事务回查方法，所以只需关注本地事务的执行状态便可。
RocketMQ提供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中间件实现可靠消息最终一致性分布式事务，模拟两个帐户的转帐交易过程。
两个帐户在分别在不一样的银行（张三在bank一、李四在bank2），bank一、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.三、spring-cloud-Greenwich.RELEASE 微服务及数据库的关系 :
dtx/dtx-txmsg-demo/dtx-txmsg-demo-bank1 银行1，操做张三帐户， 链接数据库bank1 dtx/dtx-txmsg-demo/dtx-txmsg-demo-bank2 银行2，操做李四帐户，链接数据库bank2
本示例程序技术架构以下 ：
在这里插入图片描述
交互流程以下 ：
一、Bank1向MQ Server发送转帐消息；
二、Bank1执行本地事务，扣减金额；
三、Bank2接收消息，执行本地事务，添加金额。架构

5.3.3. 建立数据库

建立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库，并导入如下表结构和数据(包含李四帐户)app

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);

在bank一、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服务器
下载地址 ：http://mirrors.tuna.tsinghua.edu.cn/apache/rocketmq/4.5.0/rocketmq-all-4.5.0-bin- release.zip
（2）解压并启动
启动nameserver ：

set ROCKETMQ_HOME=[rocketmq服务端解压路径] 
 start [rocketmq服务端解压路径]/bin/mqnamesrv.cmd

启动broker：

set ROCKETMQ_HOME=[rocketmq服务端解压路径]
start [rocketmq服务端解压路径]/bin/mqbroker.cmd ‐n 127.0.0.1:9876 autoCreateTopicEnable=true

3.3.5 工程概述

（1）父工程maven依赖说明
在父工程中指定来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>org.apache.rocketmq</groupId> 
	<artifactId>rocketmq‐spring‐boot‐starter</artifactId> 
	<version>2.0.2</version>
</dependency>

（2）配置rocketMQ
application-local.properties中配置rocketMQ nameServer地址及生产组 ：

rocketmq.producer.group = producer_bank2 
 rocketmq.name‐server = 127.0.0.1:9876

3.3.6 dtx-txmsg-demo-bank1

dtx-txmsg-demo-bank1实现以下功能：
一、张三扣减金额，提交本地事务。
二、向MQ发送转帐消息。
2）Dao

@Mapper
@Component
public interface AccountInfoDao {
	@Update("update account_info set account_balance=account_balance+#{amount} where account_no=# {accountNo}")
	int updateAccountBalance(@Param("accountNo") String accountNo, @Param("amount") Double amount);
	@Select("select count(1) from de_duplication where tx_no = #{txNo}") int isExistTx(String txNo);
	@Insert("insert into de_duplication values(#{txNo},now());") int addTx(String txNo);
}

3）AccountInfoService

@Service
@Slf4j
public class AccountInfoServiceImpl implements AccountInfoService {
	@Resource
	private RocketMQTemplate rocketMQTemplate;
	@Autowired
	private AccountInfoDao accountInfoDao;
	/** * 更新账号余额‐发送消息 * producer向MQ Server发送消息 * * @param accountChangeEvent */
	@Override
	public void sendUpdateAccountBalance(AccountChangeEvent accountChangeEvent) {
	//构建消息体
	JSONObject jsonObject = new JSONObject(); jsonObject.put("accountChange",accountChangeEvent); Message<String> message =
	MessageBuilder.withPayload(jsonObject.toJSONString()).build(); TransactionSendResult sendResult =
	rocketMQTemplate.sendMessageInTransaction("producer_group_txmsg_bank1", "topic_txmsg", message, null);
	log.info("send transcation message body={},result= {}",message.getPayload(),sendResult.getSendStatus());
	}	
	/** * 更新账号余额‐本地事务 * producer发送消息完成后接收到MQ Server的回应即开始执行本地事务 * * @param accountChangeEvent */
	@Transactional
	@Override
	public void doUpdateAccountBalance(AccountChangeEvent accountChangeEvent) {
	log.info("开始更新本地事务，事务号:{}",accountChangeEvent.getTxNo()); accountInfoDao.updateAccountBalance(accountChangeEvent.getAccountNo(),accountChangeEvent.getAmount() * ‐1);
		} 
	}
	//为幂等做准备 accountInfoDao.addTx(accountChangeEvent.getTxNo()); if(accountChangeEvent.getAmount() == 2){
	throw new RuntimeException("bank1更新本地事务时抛出异常"); log.info("结束更新本地事务，事务号:{}",accountChangeEvent.getTxNo());
		}
}

4）RocketMQLocalTransactionListener
编写RocketMQLocalTransactionListener接口实现类，实现执行本地事务和事务回查两个方法。

@Component
@Slf4j
@RocketMQTransactionListener(txProducerGroup = "producer_group_txmsg_bank1") public class ProducerTxmsgListener implements RocketMQLocalTransactionListener {
	@Autowired
	AccountInfoService accountInfoService;
	@Autowired
	AccountInfoDao accountInfoDao;
	//消息发送成功回调此方法，此方法执行本地事务
	@Override
	@Transactional
	public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object arg) {
	//解析消息内容 
	try {
		String jsonString = new String((byte[]) message.getPayload()); 
		JSONObject jsonObject = JSONObject.parseObject(jsonString); AccountChangeEvent accountChangeEvent =
		JSONObject.parseObject(jsonObject.getString("accountChange"), AccountChangeEvent.class); //扣除金额
		accountInfoService.doUpdateAccountBalance(accountChangeEvent);
		return RocketMQLocalTransactionState.COMMIT; 
	} catch (Exception e) {
	log.error("executeLocalTransaction 事务执行失败",e); e.printStackTrace();
	return RocketMQLocalTransactionState.ROLLBACK;
		} 
	}
	//此方法检查事务执行状态
	@Override
	public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
		RocketMQLocalTransactionState state;
		final JSONObject jsonObject = JSON.parseObject(new String((byte[]) message.getPayload()));
		AccountChangeEvent accountChangeEvent = JSONObject.parseObject(jsonObject.getString("accountChange"),AccountChangeEvent.class);
		//事务id
		String txNo = accountChangeEvent.getTxNo();
		int isexistTx = accountInfoDao.isExistTx(txNo);
		log.info("回查事务，事务号: {} 结果: {}", accountChangeEvent.getTxNo(),isexistTx); if(isexistTx>0){
		state= RocketMQLocalTransactionState.COMMIT; 
		}else{
		state= RocketMQLocalTransactionState.UNKNOWN; 
		}
	 	return state;
		} 
}

4）Controller

@RestController
@Slf4j
public class AccountInfoController {
@Autowired
private AccountInfoService accountInfoService;
@GetMapping(value = "/transfer")
public String transfer(@RequestParam("accountNo")String accountNo,@RequestParam("amount") Double amount){
	String tx_no = UUID.randomUUID().toString();
	AccountChangeEvent accountChangeEvent = new AccountChangeEvent(accountNo,amount,tx_no);
	accountInfoService.sendUpdateAccountBalance(accountChangeEvent);
	return "转帐成功"; 
	}
}

3.3.7 dtx-txmsg-demo-bank2

dtx-txmsg-demo-bank2须要实现以下功能 ：
一、监听MQ，接收消息。
二、接收到消息增长帐户金额。
1）Service
注意为避免消息重复发送，这里须要实现幂等。

@Service
@Slf4j
public class AccountInfoServiceImpl implements AccountInfoService {
	@Autowired
	AccountInfoDao accountInfoDao;
	/** * 消费消息，更新本地事务，添加金额 * @param accountChangeEvent */
	@Override
	@Transactional
	public void addAccountInfoBalance(AccountChangeEvent accountChangeEvent) {
		log.info("bank2更新本地帐号，帐号:{},金额: {}",accountChangeEvent.getAccountNo(),accountChangeEvent.getAmount());
		//幂等校验
		int existTx = accountInfoDao.isExistTx(accountChangeEvent.getTxNo());			if(existTx<=0){
		//执行更新 accountInfoDao.updateAccountBalance(accountChangeEvent.getAccountNo(),accountChangeEvent.getAmount());
			}
		} 
	}
	//添加事务记录
	accountInfoDao.addTx(accountChangeEvent.getTxNo()); 
	log.info("更新本地事务执行成功，本次事务号: {}", accountChangeEvent.getTxNo());
	}else{
	log.info("更新本地事务执行失败，本次事务号: {}", accountChangeEvent.getTxNo());
			}
		} 
}

2）MQ监听类

@Component
@RocketMQMessageListener(topic = "topic_txmsg",consumerGroup = "consumer_txmsg_group_bank2") 
@Slf4j
public class TxmsgConsumer implements RocketMQListener<String> {
	@Autowired
	AccountInfoService accountInfoService;
	@Override
	public void onMessage(String s) {
		log.info("开始消费消息:{}",s);
		//解析消息为对象
		final JSONObject jsonObject = JSON.parseObject(s); AccountChangeEvent accountChangeEvent =
		JSONObject.parseObject(jsonObject.getString("accountChange"),AccountChangeEvent.class);
		//调用service增长帐号金额 accountChangeEvent.setAccountNo("2"); accountInfoService.addAccountInfoBalance(accountChangeEvent);
		} 
}

5.3.8 测试场景

  • bank1本地事务失败，则bank1不发送转帐消息。
  • bank2接收转帐消息失败，会进行重试发送消息。
  • bank2屡次消费同一个消息，实现幂等。

5.4 小结

可靠消息最终一致性就是保证消息从生产方通过消息中间件传递到消费方的一致性，本案例使用了RocketMQ做为消息中间件，RocketMQ主要解决了两个功能 ：一、本地事务与消息发送的原子性问题。二、事务参与方接收消息的可靠性。可靠消息最终一致性事务适合执行周期长且实时性要求不高的场景。引入消息机制后，同步的事务操做变为基于消息执行的异步操做，避免了分布式事务中的同步阻塞操做的影响，并实现了两个服务的解耦。

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程