我说分布式事务之消息最终一致性事务（一）：原理及实现

来源：https://0x9.me/YgaUc

在之前的文章中，我们已经学习总结了分布式事务的两种解决方案。

我说分布式事务之TCC
我说分布式事务之最大努力通知型事务

本文我们将学习到另一种常见的柔性事务解决方案：消息一致性事务方案。

对于TCC型事务，跨系统的调用均是基于服务间的直接调用，即很大程度上是同步调用。基于TCC方案能够保证主子事务同时成功，同时失败。

但实际开发中，由于多方面的考虑，我们会将服务拆分为异步方式，一般是基于MQ进行服务间的解耦，服务发起方执行完本地业务操作后发送一条消息给到消息中间件（比如：RocketMQ、RabbitMQ、Kafka、ActiveMQ等），被动方服务从MQ中消费该消息并进行业务处理，从而形成业务上的闭环。

这种场景下，我们还是希望异步的多个业务操作同时成功，同时失败，基于TCC的同步型事务解决方案就不可行了，这时就需要祭出可靠消息最终一致性方案。

实现可靠消息服务

首先按照惯例我们先看一下该方案的简略的结构图，如下

核心流程1：上游投递消息

基于上图，我们分析一下如果要实现一个可靠消息一致性的服务，需要实现哪些功能，要走哪些流程。

调用开始，业务主动方（之后称为主动方）预先发送一条消息到消息服务（图中中间的部分）中，消息中包含后续的业务操作所必须的业务参数，消息服务接收到该消息后存储消息到消息存储中，并设置消息状态为 “待确认”。

如果消息存储失败则直接返回消息持久化失败，本次业务操作结束。

当主动方接收到消息存储结果后，开始执行本地的业务操作，根据本地事务提交的结果，调用消息服务的接口。这里分为两种状态：

如果本地事务执行成功，就调用消息服务确认消息状态，更新为 待发送。
如果本地事务执行失败，就调用消息服务删除消息（一般是逻辑删除，更新消息状态为 已回滚）

当状态为第1种，消息服务就将该消息发送到MQ中，并更新消息状态为 “已发送”。

注意：对于消息状态的更新和投递消息到MQ中间件的操作应在同一个方法中，并开启本地事务。为什么要这么做呢？

因为我们的目的是：保证消息发送和本地事务同时成功同时失败。

这里还是有两种情况：

如果更新消息状态失败，则应当抛出异常回滚事务，不投递消息到MQ中。
如果投递MQ失败，（需要捕获异常），需要主动抛出异常触发本地事务回滚。
1.2要同时成功同时失败

当状态是第2种，即本地事务执行失败

业务主动方需要调用可靠消息事务的删除消息操作，消息服务从消息持久化存储中删除该消息（设置消息状态为已回滚）

核心流程2：被动方应用接收消息

被动方服务订阅主题后只需要等待MQ投递消息即可。

当消息投递，被动方服务消费该消息并执行本地业务操作，当本地业务执行成功，被动方服务调用消息服务，返回本地业务执行成功。

可靠消息服务根据业务唯一参数（订单号结合消息id）设置消息状态为 “已完成”

整个过程中，作为被动方服务需要尽最大努力将业务向最终状态推进，最终成功或者失败并通知消息服务置消息状态为完成的终态。

1.如何保证消息不丢失–即保证消息可靠投递

这里分为多种情况进行讨论。

开始阶段，主动方应用提交 待确认 消息时出错，此时主动方会直接感知到提交失败，业务直接返回失败，不处理后续的流程

主动方应用执行完成本地事务之后，通知可靠消息服务确认或者删除消息阶段，出了问题：例如通知可靠消息服务失败、本地业务执行异常、可靠消息接收到提交请求后投递消息到MQ中失败等问题，如何解决？

这类情况即出现业务卡在中间态，其实没关系，因为此时消息持久化状态会一直处于 “待确认” 状态。

对于这种情况，我们只需要在可靠消息服务后台开启一个定时任务，定时扫描 “待确认” 状态的中间状态消息，当消息处于 “待确认”状态，表明主动方应用已经开始执行本地业务操作，但业务状态未知，因此我们需要对主动方本地业务执行进行回查操作。

这个阶段我们要在主动方应用中暴露一个回调查询接口，可靠消息服务会调用该接口，根据消息中的业务参数回查本地事务执行状态。如果主动方业务返回执行成功，则表明当前消息可以投递，此时可靠消息服务更新消息状态为 “待发送”，同时投递消息到MQ，并更新消息状态为 已发送。

如果，可靠消息服务（通过回查接口）询问主动方业务执行结果，返回执行失败，那么可靠消息服务需要删除该消息（逻辑删除，设置消息状态为已回滚）。

通过上述的流程，我们可以保证可靠消息服务一定会努力尝试完成消息到MQ的投递过程，即主动方业务执行与消息发送一定同时成功，同时失败。

2.如何保证消息不丢失–业务被动方对消息100%接收成功

如果消息投递成功，但业务被动方消费消息出现问题，如：消费失败、未收到消息投递（传说中的丢消息）等，该如何处理呢？

因为“未收到消息投递”的情况在消息服务高可用的情况下机会不会出现，而消费失败是业务级别的异常，因此我们同样可以采用在可靠消息服务后台起定时任务的方式，检查消息状态。

对长时间处于 “已发送” 未变更状态为 “已完成” 的消息进行重新投递操作，这个扫描的时间我们要根据业务执行时间自行调整，比如：1min。

对这类型消息重新投递到MQ之后，MQ会推送消息给消费方重新进行业务的处理操作。这个过程要在业务层实现消费的 幂等性，保证同一条消息在多次投递之后，只会进行一次完整的业务逻辑处理。关于幂等性，可以参考这篇文章 幂等的实现方案, 后续我也会针对业务层的幂等性写一篇文章出来。

整个流程中，从消息的发送，到消息的消费阶段都能保证消息与本地事务执行状态一致，即使上下游会有短暂的状态不一致，在经过一个处理的时间窗口之后，在全局上，数据能够实现最终一致性。

阶段小结

在上文，我从原理及业务上，分析了如何基于已有的消息中间件实现可靠消息服务，并实现基于可靠消息的最终一致性。

整个流程中，我们能保证：

业务主动方本地事务提交失败，业务被动方不会收到消息的投递。
只要业务主动方本地事务执行成功，那么消息服务一定会投递消息给下游的业务被动方，并最终保证业务被动方一定能成功消费该消息（消费成功或失败，即最终一定会有一个最终态）。

这个机制就是基于消息中间件的异步流程中的最终一致性保证方案。

·END·

近期热文：

Spring Cloud Stream 学习小清单
疑案追踪：Spring Boot内存泄露排查记
Spring Cloud Stream 使用延迟消息实现定时任务（RabbitMQ）
了解Java中的内存泄漏
Git 常用命令清单，掌握这些，轻松驾驭版本管理
优先级队列（头条面试题）
来谈下高并发和分布式中的幂等处理
你应该知道的7个写出更好的 Java 代码的技巧
百亿数据量下，掌握这些Redis技巧你就能Hold全场
深入聊一聊 Spring AOP 实现机制
不可错过的CMS学习笔记
可能是最全面的G1学习笔记

看完，赶紧点个“好看”鸭

点鸭点鸭

↓↓↓↓