分布式事务方案

清疚 2023-05-28 12:44 116阅读 0赞

常见的分布式事务方案一共有五种, 下面我们来一起了解下.

### 两阶段提交 ###

两阶段提交（2 Phase Commit简称2PC）是强一致性方案, 它有两个角色: 一个协调者（coordinator）和若干参与者（participant）.

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdGlhbnhpYW5nX2thb2xh_size_16_color_FFFFFF_t_70_pic_center]

*  Prepare阶段  
    协调者向所有参与者发送prepare命令, 执行事务操作, 参与者执行成功后返回应答. (命令写入redo日志后才算执行成功)
 *  Commit阶段  
    如果所有参与者都执行成功, 则协调者向所有参与者发送commit命令, 执行事务提交操作. 只有有一个参与者没有执行成功, 则协调者向所有参与者发送abort命令, 执行事务回滚操作.

> 如果在Commit阶段有参与者的commit操作执行失败, 协调者会不断重试.

这种分布式事务方案会导致一个事务的时间跨度太大, 数据锁的时间太长, 导致吞吐量严重下降. 可以通过JTA(Java Transaction API)实现.

### TCC ###

TCC是强一致性方案, 它包含三个阶段: Try, Confirm, Cancel.

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdGlhbnhpYW5nX2thb2xh_size_16_color_FFFFFF_t_70_pic_center 1]

*  Try阶段  
    所有系统执行初步操作, 如订单状态修改为PROCESSING, 可销售库存数 - 1, 冻结库存数 + 1, 预增加积分数 + 10
 *  Confirm阶段  
    当所有系统的初步操作都执行成功后, 再执行确认操作, 如订单状态修改为SUCCESS, 冻结库存数 - 1, 积分 + 10, 预增加积分数 - 10
 *  Cancel阶段  
    只要有一个系统的初步操作没执行成功, 就执行取消操作, 如订单状态修改为CLOSE, 可销售库存数 + 1, 冻结库存数 - 1, 预增加积分数 -10.

> 如果Confirm阶段和Cancel阶段有系统执行失败, 需要不断的重试.

TCC常用的开源框架有[tcc-transaction][]和[ByteTCC][]

### 本地消息表 ###

本地消息表是eBay提出的一种分布式事务方案, 该方案的核心思想在于分布式系统在处理任务时通过消息日志的方式来异步执行, 消息日志储存在消息表中, 然后通过定时任务自动重试. 本地消息表是最终一致性方案.

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdGlhbnhpYW5nX2thb2xh_size_16_color_FFFFFF_t_70_pic_center 2]

*  订单系统的数据库中有一个消息表(和订单表在同一个数据库中).
 *  订单系统收到支付成功的通知后, 在同一个事务中修改订单状态和插入消息表(执行本地事务).
 *  订单系统中存在一个定时任务, 不断地将消息表中未处理的消息发送给Kafka.
 *  库存系统和积分系统从Kafka中获取消息进行消费(注意保证幂等性).
 *  库存系统和积分系统消费成功后, 调用订单系统提供的接口, 将消息表中对应的消息标记为已处理.

本地消息表存在的目的就是确保表操作和发送消息的事务性, 如果没有本地消息表, 业务逻辑就是修改订单状态后发送消息到Kafka, 但因为这两个操作无法保证事务性, 所以可能会导致修改了订单状态, 消息却没有发送到Kafka.

### 可靠消息一致性方案 ###

可靠消息实际上是对本地消息表的一种优化方案, 我们知道本地消息表存在的目的就是确保表操作和发送消息的事务性, 如果有一种消息队列能确保这种事务性, 那就可以取消本地消息表的设计了. 我们使用RocketMQ的事务消息来实现可靠消息分布式事务方案, 这是一种最终一致性方案.

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdGlhbnhpYW5nX2thb2xh_size_16_color_FFFFFF_t_70_pic_center 3]

*  订单系统发送一个prepare消息到RocketMQ, 消息此时不可被消费.
 *  执行本地事务, 即修改订单状态
 *  如果订单状态修改成功, 则commit消息, 消息此时可被消费.
 *  如果订单状态修改失败, 则rollback消息, 消息此时被删除.
 *  为了解决commit/rollback命令发送失败的问题, RocketMQ会定时扫描所有prepare消息(默认1分钟), 然后询问发送方消息应该commit还是rollback. 此时发送方根据订单状态选择commit或rollback.
 *  库存系统和积分系统从RocketMQ中获取消息进行消费(注意保证幂等性).

伪代码实例:

/**
     * 本地事务执行监听器
     */
    public class TransactionListenerImpl implements TransactionListener{
    	private CountDownLatch latch;
    
    	public TransactionListenerImpl(CountDownLatch latch) {
    		this.latch = latch;
    	}
    
    	/**
    	 * 执行本地事务, 即修改订单状态
    	 * @param message
    	 * @param arg
    	 * @return
    	 */
    	@Override
    	public LocalTransactionState executeLocalTransaction(Message message, Object arg) {
    		/*
    			1. 修改订单状态
    			2. 如果修改成功, 则返回COMMIT_MESSAGE
    			3. 如果修改失败, 则返回ROLLBACK_MESSAGE
    		 */
    		latch.countDown();
    		return LocalTransactionState.COMMIT_MESSAGE;
    	}
    
    	/**
    	 *  RocketMQ回调检查
    	 * @param msg
    	 * @return
    	 */
    	@Override
    	public LocalTransactionState checkLocalTransaction(MessageExt msg) {
    		/*
    			1. 查询订单状态
    			2. 如果订单状态为SUCCESS, 则返回COMMIT_MESSAGE
    			3. 如果订单状态为WAIT, 则返回ROLLBACK_MESSAGE
    		 */
    		latch.countDown();
    		return LocalTransactionState.ROLLBACK_MESSAGE;
    	}
    }
    
    public class RocketMQTest {
    	private static CountDownLatch latch = new CountDownLatch(1);
    
    	public static void main(String[] args) throws Exception {
    		//本地事务执行监听器
    		TransactionListener transactionListener = new TransactionListenerImpl(latch);
    
    		//创建producer
    		TransactionMQProducer producer = new TransactionMQProducer();
    		producer.setTransactionListener(transactionListener);
    		producer.start();
    
    		//发送事务消息
    		producer.sendMessageInTransaction(new Message(), null);
    
    		latch.await();
    		producer.shutdown();
    	}
    }

### 最大努力通知 ###

上面的这几种分布式事务方案都是处理应用内的分布式事务问题, 而最大努力通知是用来处理跨应用的分布式事务, 比如对支付宝来说, 扣费成功后要通知商户.

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdGlhbnhpYW5nX2thb2xh_size_16_color_FFFFFF_t_70_pic_center 4]

*  商户提供一个接收通知的接口.
 *  支付宝扣费成功后通知商户, 如果没有收到商户"确认接收"的回复, 则不断重发.(通知次数有限, 且逐步拉大通知间隔)
 *  支付宝提供一个让商户主动查询订单是否支付成功的接口
 *  商户除了接收异步通知外, 还可以主动查询订单是否支付成功.

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdGlhbnhpYW5nX2thb2xh_size_16_color_FFFFFF_t_70_pic_center]: /images/20230528/6c8c9f553d234cb2800cfb7ca58421ac.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdGlhbnhpYW5nX2thb2xh_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20230528/3ff95987dbc0400c9c17d1c849c10624.png
[tcc-transaction]: https://github.com/changmingxie/tcc-transaction
[ByteTCC]: https://github.com/liuyangming/ByteTCC
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdGlhbnhpYW5nX2thb2xh_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20230528/5e87d72ae02443c8b2aaf88f8278a596.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdGlhbnhpYW5nX2thb2xh_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20230528/4e3d200d51ed498aa44a87e9e7d37fd3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdGlhbnhpYW5nX2thb2xh_size_16_color_FFFFFF_t_70_pic_center 4]: /images/20230528/ac33793ebf6549f79204e558ccba9772.png