分布式事务理解及解决方案

小灰灰 2022-05-14 08:45 233阅读 0赞

## 一 描述 ##

相对于单机事务的ACID原则，随着分布式微服务的普及，单机事务模式已经不能满足需求，在这里探讨一下不同服务之间的事务一致性解决方案

# #

## 二 前言 ##

阿里2017云栖大会《[破解世界性技术难题！GTS让分布式事务简单高效][GTS]》中，阿里声称提出了一种破解**世界性难题**之分布式事务的终极解决方案，无论是可靠性、还是处理速率都领先于市面上所有的技术。但令人遗憾的是一来项目**未开源**，二来还必须**依赖阿里云**的分布式数据库。毕竟，**吃饭的家伙可不能轻易示人嘛**。

虽然如此，但《世界难题...》一文中对事务还是归纳的还是蛮到位的:“一个看似简单的功能，内部可能需要调用多个“服务”并操作多个数据库或分片来实现，单一技术手段和解决方案已无法满足这些复杂应用场景。因此，分布式系统架构中分布式事务是一个绕不过去的挑战。

什么是分布式事务？简单的说，就是一次大操作由不同小操作组成，这些小操作**分布在不同服务器上，分布式事务需要保证这些小操作要么全部成功，要么全部失败**。”

举个栗子：

你上Taobao买东西，需要先扣钱，然后商品库存-1吧。但扣款和库存分别属于两个服务，这两个服务中间要经过网络、网关、主机等一系列中间层，万一任何一个地方出了问题，比如网络抖动、突发异常等待，都会导致不一致，比如扣款成功了，但是库存没-1，就会出现超卖的现象，而这就是分布式事务需要解决的问题

## 三 2阶段提交(2PC, 3PC等) ##

2阶段提交是分布式事务传统解决方案，先进为止还广泛存在。当一个事务跨越多个节点时，为了保持事务`ACID`特性，需要引入一个作为**协调者**来统一掌控所有节点(称作参与者)的操作结果并最终指示这些节点是否要把操作结果进行真正的提交(比如将更新后的数据写入磁盘等等)。因此，二阶段提交的算法思路可以概括为：**参与者**将操作成败通知协调者，再由协调者根据所有参与者的反馈情报决定各参与者是否要提交操作还是中止操作。

**以开会为例**

甲乙丙丁四人要组织一个会议，需要确定会议时间，不妨设甲是协调者，乙丙丁是参与者。

**投票阶段**

1.  甲发邮件给乙丙丁，周二十点开会是否有时间；
2.  甲回复有时间；
3.  乙回复有时间；
4.  丙迟迟不回复，此时对于这个活动，甲乙丙均处于阻塞状态，算法无法继续进行；
5.  丙回复有时间（或者没有时间）；

**提交阶段**

1.  协调者甲将收集到的结果反馈给乙丙丁（什么时候反馈，以及反馈结果如何，在此例中取决与丙的时间与决定）；
2.  乙收到；
3.  丙收到；
4.  丁收到；

不仅要锁住参与者的所有资源，而且要锁住协调者资源，开销大。一句话总结就是：2PC效率很低，对高并发很不友好。

引用`《世界性难题...》`一文原话 "国外具有几十年历史和技术沉淀的基于XA模型的商用分布式事务产品，在相同软硬件条件下，开启分布式事务后吞吐经常有数量级的下降。"

此外还有三阶段提交

![clipboard.png][]

大家有兴趣的不妨研究下

## 四 柔性事务 ##

所谓**柔性事务**是相对强制锁表的刚性事务而言。流程入下：服务器A的事务如果执行顺利，那么事务A就先行提交，如果事务B也执行顺利，则事务B也提交，整个事务就算完成。但是如果事务B执行失败，事务B本身回滚，这时事务A已经被提交，所以需要执行一个补偿操作，将已经提交的事务A执行的操作作反操作，恢复到未执行前事务A的状态。

**缺点**是业务侵入性太强，还要补偿操作，缺乏普遍性，没法大规模推广。

## 五 消息最终一致性解决方案之RocketMQ ##

目前基于消息队列的解决方案有阿里的`RocketMQ`，它实现了`半消息`的解决方案,有点类似于[Paxos算法][Paxos]，具体流程如下

**第一阶段：上游应用执行业务并发送MQ消息**

![clipboard.png][clipboard.png 1]

1.  上游应用发送待确认消息到可靠消息系统
2.  可靠消息系统保存待确认消息并返回
3.  上游应用执行本地业务
4.  上游应用通知可靠消息系统确认业务已执行并发送消息。

可靠消息系统修改消息状态为发送状态并将消息投递到 MQ 中间件

**第二阶段：下游应用监听 MQ 消息并执行业务**

下游应用监听 MQ 消息并执行业务，并且将消息的消费结果通知可靠消息服务。

![clipboard.png][clipboard.png 2]

1.  下游应用监听 MQ 消息组件并获取消息
2.  下游应用根据 MQ 消息体信息处理本地业务
3.  下游应用向 MQ
4.  确认消息被消费
5.  下游应用通知可靠消息系统消息被成功消费，可靠消息将该消息状态更改为已完成

`RocketMQ`貌似是一种先进的实现方案了，但问题是`缺乏文档`，无论是在Apache项目主页，还是在阿里的页面上，最多只告诉你如何用，而原理性或者指导性的东西非常缺乏。

当然，如果你在阿里云上专门购买了`RocketMQ`服务，想必是另当别论了。但如果你试图在自己的服务环境中部署和使用，想必要历经相当大的学习曲线。`毕竟是人家吃饭的家伙嘛`

## 六 消息最终一致性解决方案之RabbitMQ实现 ##

`RabbitMQ`遵循了`AMQP规范`，用消息确认机制来保证：**只要消息发送，就能确保被消费者消费**来做到了消息最终一致性。而且开源，文档还异常丰富，貌似是实现分布式事务的良好载体

### 6.1 RabbitMQ消息确认机制 ###

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

rabbitmq的整个发送过程如下

1. 生产者发送消息到消息服务

2. 如果消息落地持久化完成，则返回一个标志给生产者。生产者拿到这个确认后，才能放心的说消息终于成功发到消息服务了。否则进入异常处理流程。

rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
        if (!ack) {
            //try to resend msg
        } else {
            //delete msg in db
            }
        });

3. 消息服务将消息发送给消费者

4. 消费者接受并处理消息，如果处理成功则手动确认。当消息服务拿到这个确认后，才放心的说终于消费完成了。否则重发，或者进入异常处理。

final Consumer consumer = new DefaultConsumer(channel) {
          @Override
          public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        String message = new String(body, "UTF-8");
    
        System.out.println(" [x] Received '" + message + "'");
        try {
          doWork(message);
        } finally {
           //确认收到消息
          channel.basicAck(envelope.getDeliveryTag(), false);
            }
          }
        };

### 6.2 异常 ###

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

我们来看看可能发送异常的四种

**1. 直接无法到达消息服务**

网络断了，抛出异常，业务直接回滚即可。如果出现`connection closed`错误，直接增加 `connection`数即可

connectionFactory.setChannelCacheSize(100);

**2. 消息已经到达服务器，但返回的时候出现异常**

`rabbitmq`提供了确认ack机制，可以用来确认消息是否有返回。因此我们可以在发送前在db中(内存或关系型数据库)先存一下消息，如果ack异常则进行重发

/**confirmcallback用来确认消息是否有送达消息队列*/     
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
        if (!ack) {
            //try to resend msg
        } else {
            //delete msg in db
        }
        });
         /**若消息找不到对应的Exchange会先触发returncallback */
        rabbitTemplate.setReturnCallback((message, replyCode, replyText, tmpExchange, tmpRoutingKey) -> {
            try {
                Thread.sleep(Constants.ONE_SECOND);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        
            log.info("send message failed: " + replyCode + " " + replyText);
            rabbitTemplate.send(message);
        });

**3. 消息送达后，消息服务自己挂了**

如果设置了消息持久化，那么`ack= true`是在消息持久化完成后，就是存到硬盘上之后再发送的，确保消息已经存在硬盘上，万一消息服务挂了，消息服务恢复是能够再重发消息

**4. 未送达消费者**

消息服务收到消息后，消息会处于"UNACK"的状态，直到客户端确认消息

channel.basicQos(1); // accept only one unack-ed message at a time (see below)
        final Consumer consumer = new DefaultConsumer(channel) {
          @Override
          public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
            String message = new String(body, "UTF-8");
    
        System.out.println(" [x] Received '" + message + "'");
        try {
          doWork(message);
        } finally {
           //确认收到消息
          channel.basicAck(envelope.getDeliveryTag(), false);
        }
          }
        };
        boolean autoAck = false;
        channel.basicConsume(TASK_QUEUE_NAME, autoAck, consumer);

**5. 确认消息丢失**

消息返回时假设确认消息丢失了，那么消息服务会重发消息。注意，如果你设置了`autoAck= false`，但又没应答`channel.baskAck`也没有应答`channel.baskNack`，那么会导致非常严重的错误：消息队列会被堵塞住，所以，无论如何都必须应答

**6. 消费者业务处理异常**

消息监听接受消息并处理，假设抛异常了，第一阶段事物已经完成，如果要配置回滚则过于麻烦，即使做事务补偿也可能事务补偿失效的情况，所以这里可以做一个重复执行，比如`guava`的`retry`，设置一个指数时间来循环执行，如果n次后依然失败，发邮件、短信，用人肉来兜底。

[GTS]: http://jm.taobao.org/2017/04/13/20170413/
[clipboard.png]: /images/20220514/fb832e4e7d304a46990b9e5fff0a4ddb.png
[Paxos]: https://en.wikipedia.org/wiki/Paxos_%28computer_science%29
[clipboard.png 1]: /images/20220514/c79b6878bf7f4d3f805ebc7085ad3b76.png
[clipboard.png 2]: /images/20220514/ae8dbffcaf5f45be997f05db1225a5e6.png