分布式事务方案对比

灰太狼 2022-12-29 14:29 148阅读 0赞

**目录**

2PC(Prepare +Commit )

过程原理

缺点

3PC

过程原理

RocketMQ消息事务

1.发送事务消息

2.RocketMQLocalTransactionListener 接口（本地事务执行和消息事务提交，消息事务回查补偿提交）

3.本地事务类

--------------------

单体应用架构向微服务的演变随之带来的分布式事务这样的问题。分布式事务其实是要对多个微服务本地事务进行管理实现一致性。

实现分布式事务的方案比较多，常见的比如基于 XA 协议的 **2PC**和改进后的**3PC**，还有基于业务层的 **TCC**，基于**事务消息 + 消息表**实现的最终一致性方案，再有就是阿里的**Seata**中间件，下边看看各个方案的优缺点。

## 2PC(Prepare +Commit ) ##

### 过程原理 ###

**两段提交**：第一阶段，准备阶段（Prepare phase）(投票阶段) ；第二阶段，提交阶段（commit phase）。

**1. 准备阶段（Prepare phase)**

事务管理器给每个参与者发送Prepare消息，每个数据库参与者在本地执行事务，并写本地的Undo/Redo日志，此时事务没有提交。（Undo日志是记录修改前的数据，用于数据库回滚，Redo日志是记录修改后的数据，用于提交事务后写入数据文件）

**2. 提交阶段（commit phase)**

如果事务管理器收到了参与者的执行失败或者超时消息时，直接给每个参与者发送回滚(Rollback)消息；否则，发送提交(Commit)消息；参与者根据事务管理器的指令执行提交或者回滚操作，并释放事务处理过程中使用的锁资源。注意:必须在最后阶段释放锁资源。

**拿一个用户下单的例子来分析：**包含生成订单和扣减库存两个事务操作对应两个服务（订单和库存）。假设事务发起者是在订单服务内，发起后TM事务管理器会给订单和库存服务发送Prepare消息，两个服务中的RM资源管理器会执行事务操作但是不会commit，**且此时两个资源处于锁定状态**。TM得到双方的回复如果都是成功，TM会发起第二次提交commit，这时双方执行并提交事务。如果有一个失败，就统一做回滚操作。很显然，整个过程中的操作可以保持原子性。

### **缺点** ###

二阶段提交看似能够提供原子性的操作，但它存在着缺陷，而且整个过程时同步阻塞的特别是对于长事务的操作性能很差。

**1.网络抖动导致的数据不一致**： 第二阶段中协调者向参与者发送commit命令之后，一旦此时发生网络抖动，导致一部分参与者接收到了commit请求并执行，可其他未接到commit请求的参与者无法执行事务提交。进而导致整个分布式系统出现了数据不一致。

**2.超时导致的同步阻塞问题**： 2PC中的所有的参与者节点都为事务阻塞型，**当某一个参与者节点出现通信超时，其余参与者都会被动阻塞占用资源不能释放。**

## **3PC** ##

### **过程原理** ###

3PC是2PC的改进方案，2PC 中只有协调者有超时机制，会导致上面提到的由于参与者不能及时做出回复使得被动阻塞资源无法访问的问题。3PC 则在协调者和参与者中都引入了超时机制，协调者出现故障后，参与者就不会一直阻塞。而且在第一阶段和第二阶段中又插入了一个准备阶段，保证了在最后提交阶段之前各参与节点的状态是一致的。

**CanCommit**

> 协调者向所有参与者发送CanCommit命令，询问是否可以执行事务提交操作。如果全部响应YES则进入下一个阶段。

**PreCommit**

> 协调者向所有参与者发送PreCommit命令，询问是否可以进行事务的预提交操作，参与者接收到PreCommit请求后，如参与者成功的执行了事务操作，则返回Yes响应，进入最终commit阶段。一旦参与者中有向协调者发送了No响应，或因网络造成超时，协调者没有接到参与者的响应，协调者向所有参与者发送abort请求，参与者接受abort命令执行事务的中断。

**DoCommit**

> 在前两个阶段中所有参与者的响应反馈均是YES后，协调者向参与者发送DoCommit命令正式提交事务，如协调者没有接收到参与者发送的ACK响应，会向所有参与者发送abort请求命令，执行事务的中断。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI5NTY5MTgz_size_16_color_FFFFFF_t_70][]

## RocketMQ消息事务 ##

从网上找了一张图。

MQ消息事务实现方式有很多花样。RocketMQ的实现方式看起来比其他都要合适，因为RocketMQ独有的事务回调扩展。

下面流程图的大概思路：

> 假设有两个本地事务组成当前的全局事务，在执行第一事务前，发送half消息到MQ，MQserver收到后保存消息但是不可消费状态，MQ回调发送者的回调接口，这时候在接口中我们执行本地事务，相当于把本地事务包装在当前回调接口方法事务内，如果本地事务执行成功，久提交MQ的事务，此时MQ会把消息设置为可消费状态，否则执行事务回滚，本地事务失败且消息也会被删除。
> 
> 另外MQ会回查发送者的事务状态，可以做补偿提交。
> 
> 上述流程保证了第一个本地事务与消息发送的一致性，即本地事务发送成功后消息才可消费。基于MQ的分布式事务实现的是最终一致性并不保证实时性，所以对于消费者而言只要确保收到消息完成第二个本地事务的提交就可以了。

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

### 1.发送事务消息 ###

@RestController
    @Slf4j
    public class AccountInfoController {
        @Autowired
        private AccountInfoService accountInfoService;
    
        @GetMapping(value = "/transfer")
        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 "转账成功";
        }
    }

### 2.RocketMQLocalTransactionListener 接口（本地事务执行和消息事务提交，消息事务回查补偿提交） ###

@Component
    @Slf4j
    @RocketMQTransactionListener(txProducerGroup = "producer_group_txmsg_bank1")
    public class ProducerTxmsgListener implements RocketMQLocalTransactionListener {
    
        @Autowired
        AccountInfoService accountInfoService;
    
        @Autowired
        AccountInfoDao accountInfoDao;
    
        //事务消息发送后的回调方法，当消息发送给mq成功，此方法被回调
        @Override
        @Transactional
        public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
    
            try {
                //解析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);
                //执行本地事务，扣减金额
                accountInfoService.doUpdateAccountBalance(accountChangeEvent);
                //当返回RocketMQLocalTransactionState.COMMIT，自动向mq发送commit消息，mq将消息的状态改为可消费
                return RocketMQLocalTransactionState.COMMIT;
            } catch (Exception e) {
                e.printStackTrace();
                return RocketMQLocalTransactionState.ROLLBACK;
            }
    
        }
    
        //MQ回调事务状态回查接口，查询是否扣减金额
        @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();
            log.info("事务状态回查");
            int existTx = accountInfoDao.isExistTx(txNo);
            if(existTx>0){
                return RocketMQLocalTransactionState.COMMIT;
            }else{
                return RocketMQLocalTransactionState.UNKNOWN;
            }
        }
    }

### 3.本地事务类 ###

@Service
    @Slf4j
    public class AccountInfoServiceImpl implements AccountInfoService {
    
        @Autowired
        AccountInfoDao accountInfoDao;
    
        @Autowired(required = false)
        RocketMQTemplate rocketMQTemplate;
    
    
        //向mq发送转账消息
        @Override
        public void sendUpdateAccountBalance(AccountChangeEvent accountChangeEvent) {
    
            //将accountChangeEvent转成json
            JSONObject jsonObject =new JSONObject();
            jsonObject.put("accountChange",accountChangeEvent);
            String jsonString = jsonObject.toJSONString();
            log.info(jsonString);
            //生成message类型
            Message<String> message = MessageBuilder.withPayload(jsonString).build();
            //发送一条事务消息
            /**
             * String txProducerGroup 生产组
             * String destination topic，
             * Message<?> message, 消息内容
             * Object arg 参数
             */
            rocketMQTemplate.sendMessageInTransaction("producer_group_txmsg_bank1","topic_txmsg",message,null);
    
        }
    
        //更新账户，扣减金额
        @Override
        @Transactional
        public void doUpdateAccountBalance(AccountChangeEvent accountChangeEvent) {
            //幂等判断
            if(accountInfoDao.isExistTx(accountChangeEvent.getTxNo())>0){
                return ;
            }
            //扣减金额
            accountInfoDao.updateAccountBalance(accountChangeEvent.getAccountNo(),accountChangeEvent.getAmount() * -1);
            //添加事务日志
            accountInfoDao.addTx(accountChangeEvent.getTxNo());
            if(accountChangeEvent.getAmount() == 3){
                throw new RuntimeException("人为制造异常");
            }
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI5NTY5MTgz_size_16_color_FFFFFF_t_70]: /images/20221120/7bf9307981a94a4a9d951a1b3076dfd1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDA2MjMzOQ_size_16_color_FFFFFF_t_70]: /images/20221120/c8b6285ea8a8499491afca277cb894a2.png