文章目录
  1. 1. 实现可靠消息服务
    1. 1.1. 核心流程1–上游投递消息
    2. 1.2. 核心流程2–被动方应用接收消息
    3. 1.3. 1.如何保证消息不丢失–即保证消息可靠投递
    4. 1.4. 2.如何保证消息不丢失–业务被动方对消息100%接收成功
  2. 2. 阶段小结

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

我说分布式事务之TCC

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

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

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

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

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

实现可靠消息服务

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

消息一致性方案

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

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

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

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

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

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

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

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

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

这里还是有两种情况:

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

如果消息投递成功,但业务被动方消费消息出现问题,如:消费失败、未收到消息投递(传说中的丢消息)等,该如何处理呢?

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

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

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

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

阶段小结

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

整个流程中,我们能保证:

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

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

下一篇文章,我将介绍如何基于开源框架RocketMQ实现可靠消息一致性。

文章目录
  1. 1. 实现可靠消息服务
    1. 1.1. 核心流程1–上游投递消息
    2. 1.2. 核心流程2–被动方应用接收消息
    3. 1.3. 1.如何保证消息不丢失–即保证消息可靠投递
    4. 1.4. 2.如何保证消息不丢失–业务被动方对消息100%接收成功
  2. 2. 阶段小结
Fork me on GitHub