如何能在实战中完成分布式事务

背景

在一年前我写过一篇关于分布式事务的文章：

再有人问你分布式事务，把这篇扔给他 ，在这篇文章中我详细介绍了分布式事务是什么，实现分布式事务有哪些常用的方案，但是其中的东西很多是偏于理论，很多读者对其真正在实战上的使用可能还是有点差距。所以在前几次文章的更新中，我介绍了很多关于Seata（一款由阿里开源的分布式事务框架）的内容，如果大家对Seata不是很熟悉的可以阅读下面的内容：

• 解密分布式事务框架-Seata

• 深度剖析一站式分布式事务方案Seata-Server

• 深度剖析一站式分布式事务方案Seata-Cient

Seata已经为我们提供了两种实现分布式模式：

• AT：自动模式，通过我们记录运行sql的undolog，来完成事务失败时的自动重做。

• TCC：TCC模式，这种模式弥补我们AT模式只能支持ACID数据库的场景。

大多数时候Seata已经足够了，但是很多时候不同场景下我们没办法选择Seata这类TCC框架：

• 改造困难，目前Seata支持的通信框架不多只有Dubbo和Spring-Cloud-Alibaba，如果使用的是其他框架，或者直接是简单的HTTP，甚至有些公司可能目前系统中都没有支持Trace。

• 维护成本高，Seata需要一个单独的集群去维护，一般在公司都需要分配一定的资源（人员资源，机器资源）去管理维护Seata，很多时候不可能为了几个分布式事务去花费这么大的成本，当然这一块的话未来可以上云解决。

而我最近在做一些分布式事务的事的时候也遇到了这些问题，由于一般使用分布式事务是业务方，你需要驱动做RPC组件的同事支持，并且我们并不是纯金融服务的公司，搭建一套类似Seata的分布式事务中间件也是比较耗费资源。

之前介绍的方案大多数都比较笼统，俗话说授人以鱼不如授人以渔，所以接下来我将会一步一步的教大家如何不用框架，而是我们自己去编码去实现分布式事务。

问题

为了更好的讲解如何在实战中完成分布式事务，这里直接举一个大家都熟悉的例子：用户下单的时候，可以选择三种资产，分别是储值余额，积分，券，这个场景几乎在每个应用都能看见，而这个场景在我们的后端可以映射为4个服务，如下图所示：

在这个场景下大多数人的代码基本会按照下面的写,在订单服务中有如下步骤，这里为了简单没有设置过多的订单状态：

• Step 1：创建订单状态为初始化，并检查用户所有资源是否足够

• Step 2：支付储值余额

• Step 3：支付券

• Step 4：支付金币

• Step 5：更新订单状态为已完成

差不多这里就是简简单单4行，有很多人会把这5步直接放进事务之中，也就是加上@Transactional注解，但其实加上这个注解不仅没有起到事务作用，而且还让我们的事务变成了长事务，我们这里的Step2-4都是RPC远程调用，一旦某个RPC出现了Timeout，那么我们的数据库连接会被长期持有不被释放，有可能导致我们系统雪崩。

既然这里加上事务没有用，我们可以看看会出现什么问题，如果Step2支付成功，Step3失败，那么就会导致数据不一致。其实很多人就会有侥幸心理，默认我们的Step 2-4会成功，如果出现问题我们人工修复就是了。人工修复的成本太高，你就想如果你在外面旅游突然叫你修复数据，那你是不是会气得吐血？所以我们这里一步一步的教大家如何逐渐的把这段业务逻辑优化成能保证我们数据一致的。

方法

一般来说任何一个分布式事务框架都离不开三个关键字：重做记录，重试机制，幂等。而在我们的业务中同样也离不开这三个关键字。

重做记录

我们想想我们mysql的事务回滚是依靠什么的？依靠的是undolog，我们的undolog保存了事务发生之前的数据的一个版本，那么我们发生回滚的时候直接利用这个版本的数据回滚即可。这里我们首先需要添加我们的重做记录，我们没必要叫undolog，我们再各个资源服务中需要添加一个事务记录表：

CREATE TABLE `transaction_record` (
  `orderId` int(11) unsigned NOT NULL AUTO_INCREMENT,
  `op_total` int(11) NOT NULL COMMENT '本次操作资源操作数量',
  `status` int(11) NOT NULL COMMENT '1:代表支付成功 2:代表支付取消',
  `resource_id` int(11) NOT NULL COMMENT '本次操作资源的Id',
  `user_id` int(11) NOT NULL COMMENT '本次操作资源的用户Id',
  PRIMARY KEY (`orderId`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;

在我们的分布式事务中有一个全局事务ID,而我们 orderId 就能很好的适应这个角色，这里我们每个资源的事务记录表都需要记录这个OrderId，用于和全局事务进行关联，并且我们这里直接将其作为主键，也表明了这个表中只会出现一次全局事务ID。这里的 op_total 用于记录本次操作资源的数量，用于后续回滚，哪怕不回滚我们也可以用于后续记录的查询。 status 用于记录我们当前这条记录的状态如何，这里用了两个状态，后续我们可以扩展更多的状态，解决更多的分布式事务问题。

有了这个重做记录之后我们只需要在每一次执行记录下我们的当前资源的transaction_record，在回滚的时候根据我们的OrderId将所有的资源回滚，我们优化之后代码可以如下：

        int orderId = createInitOrder();
        checkResourceEnough();
        try {
            accountService.payAccount(orderId, userId, opTotal);
            coinService.payCoin(orderId, userId, opTotal);
            couponService.payCoupon(orderId, userId, couponId);
            updateOrderStatus(orderId, PAID);
        }catch (Exception e){
            //这里进行回滚
            accountService.rollback(orderId, userId);
            coinService.rollback(orderId, userId);
            couponService.rollback(orderId, userId);
            updateOrderStatus(orderId, FAILED);
        }

这里我们将创建好的初始化订单，当作参数传递给我们的资源服务记录，最后再进行状态更新，如果发生了异常，那么我们需要进行 手动 回滚并将订单数据变为FAILED, 回滚的依据就是我们的订单Id。对于我们的支付和回滚的伪代码有如下：

    @Transactional
    void payAccount(int orderId, int userId, int opTotal){
        account.payAccount(userId, opTotal); // 实际的去我们account表扣减
        transactionRecordStorage.save(orderId, userId, opTotal, account.getId()); //保存事务记录表
    }
    @Transactional
    void rollback(int orderId, int userId){
        TransactionRecord tr = transactionRecordStorage.get(orderId); //从记录表中查询
        account.rollbackBytr(tr); // 根据记录回滚
    }

这里的版本是比较简略的，问题还比较多后面会讲优化。

重试机制

有些同学可能会问好像我们上面的代码基本能保证分布式事务了吧？的确上面的代码能保证我们在没有宕机或者其他更加严重的情况下基本上是没有问题的，但是如果出现了宕机，比如我们刚刚把account给支付完了，然后支付coin的时候我们的订单机器宕机了，没有发出去这个请求，这里就不会走到我们的手动回滚请求，所以我们的account将会永远不会回滚，又只得靠我们的人工回滚，如果你此时还在旅游，又叫你回滚，估计你会继续气晕。或者说我们再回滚的时候出现错误，怎么办？我们没有有效的手段进行针对回滚的回滚。

所以我们需要额外的重试机制来保证，首先我们需要定义什么样的数据需要重试，这里的话我们根据业务差不多一分钟能将所有的都资源都支付完，如果我们的订单状态为init 并且 创建时间超过一分钟，那么就认为发生了上述错误的事件。接下来可以通过我们的重试机制进行回滚，这里有两个常见重试机制：

• 定时任务：定时任务是我们最常见的重试机制，基本所有的分布式事务框架中也都是通过定时任务去做的，这里我们需要使用分布式的定时任务，分布式的定时任务可以使用单机任务+分布式锁 或者 直接使用开源的分布式任务中间件如elastic-job。我们在分布式任务的逻辑中每次查询我们的处于订单状态为init 并且 创建时间超过一分钟的订单，我们对其进行回滚，回滚完成之后将订单状态置为FAILED。

• 消息队列：目前我们业务上使用的是消息队列，将下单操作放入消息队列中去做，如果我们出现了各种异常，那么我们依靠消息队列的重试机制，一般来说现在当前队列进行重试，再丢给死信队列去重试。这里的逻辑就需要改一下，在我们创建订单的时候有可能订单已经存在，如果存在的话我们判断他的状态（init+1min）是否应该被直接rollback，如果是则直接1min。为什么我们选择了消息队列进行重试？ 因为我们的业务逻辑是依靠消息队列的，我们就不需要引入定时任务，直接依靠消息队列即可。

幂等

判断一个程序猿经验是否老道可以从他写代码的时候能否考虑到幂等就可以看出。很多年轻的程序员根本不会考虑幂等的存在，甚至都不知道幂等是什么。这里先解释一下幂等的概念：可以简单的认为任意多次执行所产生的影响和一次执行的影响相同。

为什么我们完成分布式事务的时候需要幂等？大家可以想想如果在执行回滚操作的时候宕机了，我们上面的重试机制就会开始工作，比如我们的券这个资源已经回滚，但是我们重试操作的时候我并不知道券已经回滚了，这个时候就再次尝试回滚券，如果没有做幂等操作会怎么办，有可能导致用户资产会多增加，这样就会对公司造成很多损失。

所以幂等在我们重试的时候非常重要，实现幂等的关键是什么？我们想让多次操作和一次操作是一样的，那么我们只需要比较第一次已经做过了，而这个标记通过什么来完成呢？这里我们可以使用我们状态机转换的手段完成标记。只有标记这里还是不够，为什么呢这里我们用个例子来说明一下，把上面的rollback简单优化一下：

    @Transactional
    void rollback(int orderId, int userId){
        TransactionRecord tr = transactionRecordStorage.get(orderId);
        if(tr.isCanceled()){
            return; //如果已经被取消了那么直接返回
        }
        //从记录表中查询
        account.rollbackBytr(tr); // 根据记录回滚
    }

上面代码我们通过判断状态如果是已经被取消了，也就是被回滚了那么我们就直接返回，这里就完成了我们所说的幂等。但是这里还有个问题是如果有两个rollback同时执行怎么办？你可能会问什么样的情况可能会有两个rollback，这里举一个场景当第一次rollback的时候请求在阻塞了，这个时候调用方已经触发超时了，然后一段时间之后第二次rollback来了，这个时候恰好第一次也不阻塞了，那么这里就会有两个rollback请求发出，当执行状态判断的时候，如果两个请求同时执行状态判断，那么都会绕过这个检查，最后用户就会退两次钱，这样的情况我们一定要避免。

那么怎么才能避免呢？聪明的同学马上就会想到使用分布式锁呀，一提到分布式锁马上想到的就是Redis加锁，ZK加锁等等，我在这篇文章也做了介绍：聊聊分布式锁，但是我们这里直接使用数据库行锁即可，也就是用下面的sql语句查询：

select * from transaction where orderId = "#{orderId}" for update;

其他的代码不变，通过这种形式我们完成了幂等。这时候有可能会有同学会问到，如果TransactionRecord不存在怎么办？因为我们重试的时候我们怎么知道他的Try是否成功，我们这里是不知道的，所以我们这里还有策略保证我们的逻辑不会出现空指针，这里有两种策略来做这个事：

• 如果为空我们直接返回即可。

• 如果为空，我们保存一条Status为已执行空回滚状态的TransactionRecord。

上面的第一个策略比较简单，但是我们这里需要选择第二个策略，为什么呢因为我们还需要预防一个事情：防悬挂，我们再说rollback幂等的时候，如果第一个rollback发生网络阻塞，那么这里我们将rollback替换成我们第一次支付的时候发生了阻塞，导致了pay在rollback之后到达我们的客户端，如果我们采用第一种方式，我们这个阻塞的Pay请求时无法感知整个事务因为rollback，然后继续pay导致我们这个pay永远得不到回滚，这就是悬挂。所以我们这里采用第二个策略，保存一条记录，我们在pay也会检查有没有这条记录，所以优化之后的代码为：

    @Transactional
    void payAccount(int orderId, int userId, int opTotal){
        TransactionRecord tr = transactionRecordStorage.getForUpdate(orderId);
        if(tr != null){
            return; //如果已经有数据了，这里直接返回
        }
        account.payAccount(userId, opTotal); // 实际的去我们account表扣减
        transactionRecordStorage.save(orderId, userId, opTotal, account.getId()); //保存事务记录表
    }
    @Transactional
    void rollback(int orderId, int userId){
         TransactionRecord tr = transactionRecordStorage.getForUpdate(orderId);
        if(tr == null){
            saveNullCancelTr(orderId, userId); //保存空回滚的记录
        }
        if(tr.isCanceled() || tr.isNullCancel()){
            return; //如果已经被取消了那么直接返回
        }
        //从记录表中查询
        account.rollbackBytr(tr); // 根据记录回滚
    }

总结

到这里我们整个构建分布式事务基本大功告成了，通过这种方式基本上以后遇到相关分布式事务的业务问题的时候都可以解决。这里我们再回顾一下我们的三个要点：

• 重试记录：通过数据记录保存。

• 重试机制：定时任务或者消息队列自带的重试。

• 幂等：通过状态机加数据库行锁。

我们只要能掌握好这三个点，其实不仅仅是对分布式事务这一块有帮助，对其他的业务同样也有很大的提升。

如果大家觉得这篇文章对你有帮助，你的关注和转发是对我最大的支持，O(∩_∩)O: