服务化带来的数据一致问题---分布式事务，事务型消息

用户7927337

发布于 2020-11-04 14:20:42

2K0

发布于 2020-11-04 14:20:42

文章被收录于专栏：二马读书二马读书

本文我们聊聊分布式事务和事务型消息的解决思路，通过阅读本文，可以理解分布式事务和事务型消息，并且能够应用到实际生产工作中。

服务化后单体系统被拆分成多个服务，各服务访问自己的数据库。而我们的一次请求操作很可能要跨多个服务，同时要操作多个数据库的数据，我们发现以前用的数据库事务不好用了。那么基于微服务的架构如何保证数据一致性呢？

好，咱们这次就盘一盘分布式事务，最终一致，补偿机制，事务型消息！

提起这些，大家可能会想到两阶段提交，XA，TCC，Saga，还有最近阿里开源的Seata（Fescar），这些概念网上一大堆文章，不过都太泛泛，不接地气，让人看了云里雾里

这次咱们以TCC分布式事务和RocketMQ事务型消息为例，做详细分享！这个弄明白了，也就清楚分布式事务，最终一致，补偿机制这些概念啦！

TCC分布式事务

TCC（Try-Confirm-Cancel）是分布式事务的一种模式，可以保证不同服务的数据最终一致。目前有不少TCC开源框架，比如Hmily，ByteTCC，TCC-Transaction （我们之前用过Hmily和公司架构组自研组件）

下面以电商下单流程为例对TCC做详细阐述

流程图如下：

基本步骤如下：

1，修改订单状态为“已支付”

2，扣减库存

3，扣减优惠券

4，通知WMS（仓储管理系统）捡货出库（异步消息）

我们先看扣减库存，更新订单状态和扣减优惠券这三步同步调用，通知WMS的异步消息会在后面的“基于消息的最终一致”部分详细阐述！

下面是伪代码。不同公司的产品逻辑会有差异，相关代码逻辑也可能会有不同，大家不用纠结代码逻辑正确性。

public void makePayment() {
   orderService.updateStatus(OrderStatus.Payed); //订单服务更新订单为已支付状态
   inventoryService.decrStock(); //库存服务扣减库存
   couponService.updateStatus(couponStatus.Used); //卡券服务更新优惠券为已使用状态      
}

看完这段代码，大家可能觉得很简单！那么有什么问题吗？

答案是肯定的。没法保证数据一致性，也就是说不能保证这几步操作全部成功或者全部失败！因为这几步操作是在分布式环境下进行的，每个操作分布在不同的服务中，不同的服务又对应不同的数据库，本地事务已经用不上了！假如第一步更新订单为“已支付”成功了，第二步扣减库存时，库存服务挂了或者网络出问题了，导致扣减库存失败。你告诉用户支付成功了，但是库存没扣减。这怎么能行！

接下来，我们来看看TCC是如何帮我们解决这个问题并保证数据最终一致的。

TCC分为两个阶段：

阶段一：Try（预留冻结相关业务资源，设置临时状态，为下个阶段做准备）

阶段二：Confirm 或者 Cancel（Confirm：对资源进行最终操作，Cancel：取消资源）

第一阶段：Try

更新订单状态：此时因为还没真正完成整个流程，订单状态不能直接改成已支付状态。可以加一个临时状态Paying，表明订单正在支付中，支付结果暂时还不清楚！
冻结库存：假设现在可销售库存stock是10，这单扣减1个库存，别直接把库存减掉，而是在表中加一个冻结字段locked_stock，locked_stock 加 1，再给stock减1，这样就相当于冻结了1个库存。两个操作放在一个本地事务里。
更新优惠券状态：优惠券加一个临时状态Inuse，表明优惠券正在使用中，具体有没有正常被使用暂时还不清楚！

第二阶段：Confirm

假如第一阶段几个try操作都成功了！既然第一阶段已经预留了库存，而且订单状态和优惠券状态也设置了临时状态，第二阶段的确认提交从业务上来说应该没什么问题了。

Confirm 阶段我们需要做下面三件事：

先将订单状态从Paying改为已支付Payed，订单状态也完成了。
再将冻结的库存恢复locked_stock减1，stock第一阶段已经减掉1是9了，到此扣减库存就真正完成了。
再将优惠券状态从Inuse改为Used，表明优惠券已经被正常使用。

第二阶段：Cancel

假如第一阶段失败了，

先将订单状态从Paying恢复为待支付UnPayed。
再将冻结的库存还回到可销售库存中，stock加1恢复成10， locked_stock减1，可以放在一个本地事务完成。
再将优惠券状态从Inuse恢复为未使用Unused。

基于Hmily框架的代码：

//订单服务
public class OrderService{

  //tcc接口
  @Hmily(confirmMethod = "confirmOrderStatus", cancelMethod = "cancelOrderStatus")
  public void makePayment() {
     更新订单状态为支付中
     冻结库存，rpc调用
     优惠券状态改为使用中，rpc调用
  }
  
  public void confirmOrderStatus() {
     更新订单状态为已支付
  }

  public void cancelOrderStatus() {
     恢复订单状态为待支付 
  }  
  
}

//库存服务
public class InventoryService {

  //tcc接口
  @Hmily(confirmMethod = "confirmDecr", cancelMethod = "cancelDecr")
  public void lockStock() {
     //防悬挂处理（下面有说明）
     if (分支事务记录表没有二阶段执行记录)
       冻结库存
     else
       return;
  }
  
  public void confirmDecr() {
     确认扣减库存
  }
  public void cancelDecr() {
     释放冻结的库存
  }  
  
}

//卡券服务
public class CouponService {

  //tcc接口
  @Hmily(confirmMethod = "confirm", cancelMethod = "cancel")
  public void handleCoupon() {
     //防悬挂处理（下面有说明）
     if (分支事务记录表没有二阶段执行记录)
       优惠券状态更新为临时状态Inuse
     else
       return;
  }
  
  public void confirm() {
     优惠券状态改为Used
  }
  public void cancel() {
     优惠券状态恢复为Unused
  }  
  
}

疑问？

问题1：有些朋友可能会问了！这些关于流程的逻辑也要手动编码吗？这也太麻烦了吧！

实际上TCC分布式事务框架帮我们把这些事都干了。比如我们前面提到的Hmily，ByteTCC，TCC-transaction 这些框架。因为try，confirm，cancel这些操作都在TCC分布式事务框架控制范围之内，所以try的各个步骤成功了或者失败了，框架本身都知道，try成功了框架就会自动执行各个服务的confirm，try失败了框架就会执行各个服务的cancel（各个服务内部的TCC分布式事务框架会互相通信）。所以我们不用关心流程，只需要关注业务代码就可以啦！

问题2：仔细想想，好像还有问题！假如confirm阶段更新订单状态成功了，但是扣减库存失败了怎么办呢？

比如网络出问题了或者库存服务（宕机，重启）出问题了。当然，分布式事务框架也会考虑这些场景，框架会记录操作日志，假如confirm阶段扣减库存失败了，框架会不断重试调用库存服务直到成功（考虑性能问题，重试次数应该有限制）。cancel过程也是一样的道理。注意，既然需要重试，我们就要保证接口的幂等性。什么？不太懂幂等性。简单说：一个操作不管请求多少次，结果都要保证一样。这里就不详细介绍啦！

再考虑一个场景，try阶段冻结库存的时候，因为是rpc远程调用，在网络拥塞等情况下，是有可能超时的。假如冻结库存时发生超时，tcc框架会回滚（cancel）整个分布式事务，回滚完成后冻结库存请求才到达参与者（库存服务）并执行，这时被冻结的库存就没办法处理（恢复）了。这种情况称之为“悬挂”，也就是说预留的资源后续无法处理。解决方案：第二阶段已经执行，第一阶段就不再执行了，可以加一个“分支事务记录表”，如果表里有相关第二阶段的执行记录，就不再执行try（上面代码有防悬挂处理）。有人可能注意到还有些小纰漏，对，加锁，分布式环境下，我们可以考虑对第二阶段执行记录的查询和插入加上分布式锁，确保万无一失。

基于消息的最终一致

还是以上面的电商下单流程为例

上图，下单流程最后一步，通知WMS捡货出库，是异步消息走消息队列。

public void makePayment() {
   orderService.updateStatus(OrderStatus.Payed); //订单服务更新订单为已支付状态
   inventoryService.decrStock(); //库存服务扣减库存
   couponService.updateStatus(couponStatus.Used); //卡券服务更新优惠券为已使用状态      
   发送MQ消息捡货出库； //发送消息通知WMS捡货出库
}

按上面代码，大家不难发现问题！如果发送捡货出库消息失败，数据就会不一致！有人说我可以在代码上加上重试逻辑和回退逻辑，发消息失败就重发，多次重试失败所有操作都回退。这样一来逻辑就会特别复杂，回退失败要考虑，而且还有可能消息已经发送成功了，但是由于网络等问题发送方没得到MQ的响应，这些问题都要考虑进来！

幸好，有些消息队列帮我们解决了这些问题。比如阿里开源的RocketMQ（目前已经是Apache开源项目），4.3.0版本开始支持事务型消息（实际上早在贡献给Apache之前曾经支持过事务消息，后来被阉割了，4.3.0版本重新开始支持事务型消息）。

先看看RocketMQ发送事务型消息的流程：

1，发送半消息（所有事务型消息都要经历确认过程，从而确定最终提交或回滚（抛弃消息），未被确认的消息称为“半消息”或者“预备消息”，“待确认消息”）

2，半消息发送成功并响应给发送方

3，执行本地事务，根据本地事务执行结果，发送提交或回滚的确认消息

4，如果确认消息丢失（网络问题或者生产者故障等问题），MQ向发送方回查执行结果

5，根据上一步骤回查结果，确定提交或者回滚（抛弃消息）

看完事务型消息发送流程，有些读者可能没有完全理解，不要紧，我们来分析一下！

问题1：假如发送方发送半消息失败怎么办？

半消息（待确认消息）是消息发送方发送的，如果失败，发送方自己是知道的并可以做相应处理。

问题2：假如发送方执行完本地事务后，发送确认消息通知MQ提交或回滚消息时失败了（网络问题，发送方重启等情况），怎么办？

没关系，当MQ发现一个消息长时间处于半消息（待确认消息）的状态，MQ会以定时任务的方式主动回查发送方并获取发送方执行结果。这样即便出现网络问题或者发送方本身的问题（重启，宕机等），MQ通过定时任务主动回查发送方基本都能确认消息最终要提交还是回滚（抛弃）。当然出于性能和半消息堆积方面的考虑，MQ本身也会有回查次数的限制。

问题3：如何保证消费一定成功呢？

RocketMQ本身有ack机制，来保证消息能够被正常消费。如果消费失败（消息订阅方出错，宕机等原因），RocketMQ会把消息重发回Broker，在某个延迟时间点后（默认10秒后）重新投递消息。

结合上面几个同步调用hmily完整代码如下：

//TransactionListener是rocketmq接口用于回调执行本地事务和状态回查
public class TransactionListenerImpl implements TransactionListener {
     //执行本地事务
     @Override
     public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
         记录orderID，消息状态键值对到共享map中，以备MQ回查消息状态使用；
         return LocalTransactionState.COMMIT_MESSAGE;
     }
 
     //回查发送者状态
     @Override
     public LocalTransactionState checkLocalTransaction(MessageExt msg) {
         String status = 从共享map中取出orderID对应的消息状态; 
         if("commit".equals(status))
           return LocalTransactionState.COMMIT_MESSAGE;
         else if("rollback".equals(status))
           return LocalTransactionState.ROLLBACK_MESSAGE;
         else
           return LocalTransactionState.UNKNOW;
     }
}

//订单服务
public class OrderService{

  //tcc接口
  @Hmily(confirmMethod = "confirmOrderStatus", cancelMethod = "cancelOrderStatus")
  public void makePayment() {
     1，更新订单状态为支付中
     2，冻结库存，rpc调用
     3，优惠券状态改为使用中，rpc调用
     4，发送半消息（待确认消息）通知WMS捡货出库 //创建producer时这册TransactionListenerImpl
  }
  
  public void confirmOrderStatus() {
     更新订单状态为已支付
  }

  public void cancelOrderStatus() {
     恢复订单状态为待支付 
  }  
  
}

//库存服务
public class InventoryService {

  //tcc接口
  @Hmily(confirmMethod = "confirmDecr", cancelMethod = "cancelDecr")
  public void lockStock() {
     //防悬挂处理
     if (分支事务记录表没有二阶段执行记录)
       冻结库存
     else
       return;
  }
  
  public void confirmDecr() {
     确认扣减库存
  }
  public void cancelDecr() {
     释放冻结的库存
  }  
  
}

//卡券服务
public class CouponService {

  //tcc接口
  @Hmily(confirmMethod = "confirm", cancelMethod = "cancel")
  public void handleCoupon() {
     //防悬挂处理
     if (分支事务记录表没有二阶段执行记录)
       优惠券状态更新为临时状态Inuse
     else
       return;
  }
  
  public void confirm() {
     优惠券状态改为Used
  }
  public void cancel() {
     优惠券状态恢复为Unused
  }  
  
}

如果执行到TransactionListenerImpl.executeLocalTransaction方法,说明半消息已经发送成功了，也说明OrderService.makePayment方法的四个步骤都执行成功了，此时tcc也到了confirm阶段，所以在TransactionListenerImpl.executeLocalTransaction方法里可以直接返回LocalTransactionState.COMMIT_MESSAGE 让 MQ提交这条消息，同时将该订单信息和对应的消息状态保存在共享map里，以备确认消息发送失败时MQ回查消息状态使用。

以上是我个人对分布式事务，事务型消息等的一些总结，公众号近期文章菜单中还有关于服务化问题的其他文章，如：服务雪崩，服务化过程数据迁移等文章，感谢关注和阅读！