摘要: spring cloud实现可靠消息一致性

1.1 目的

1. 为开发测试提供指导性文件

2. 为系统今后的扩展提供参考

3. 解决系统中消息不可达问题

1.2 范围和功能

1.3 适用读者

1. 需要发送MQ分布式系统的开发人员和测试人员

2. 可靠消息服务的开发人员和测试人员

1.4 读者须知

本服务需要提供一个sdk和数据库初始语句创建数据库表,并且对外提供可扫描的domain、mapper、service,使用的技术框架zk + mapper3 + pagehelper + feign(edas) , 使用者(上游系统、下游系统) 只需要在对应的接口上写上响应注解即可实现可靠消息, 如果不熟悉上述框架,可选择对应框架替换,比如redis替换zk,放弃mapper3和pagehelper使用传统的mybatis,使用http接口替换fein(eads)的解决办法,本文不提供替换的解决方案

1.5 参考文档

本文为分布式系统解决方案,此方案涉及 3 个模块：

1. 上游应用，执行业务并发送指令给可靠消息服务并保留消息副本。

2. 可靠消息服务和 MQ消息组件，协调上下游消息的传递，并确保上下游数据的一致性。

3. 下游应用，监听 MQ 的消息并执行自身业务并保留消息副本。

2.1 业务流程图

2.2 数据库表设计

2.2.1 可靠消息表

2.2.2 消费者确认表

2.2.3 消费者表

2.2.4 生产者表

2.2.5 发布关系表

2.2.6 消息重发记录表

暂时未设计

2.2.7 消息订阅关系表

2.2.8 消息订阅TAG关系表

2.2.9 各个子系统消息落地的消息表

3.1 上游应用执行业务并发送 MQ 消息

上游应用将本地业务执行和消息发送绑定在同一个本地事务中，保证要么本地操作成功并发送 MQ 消息，要么两步操作都失败并回滚。这里可以采用自定义切面完成,后续会有介绍。

1. 上游应用发送待确认消息到可靠消息系统。(本地消息落地)

2. 可靠消息系统保存待确认消息并返回。

3. 上游应用执行本地业务。

4. 上游应用通知可靠消息系统确认业务已执行并发送消息。

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

以上每一步都可能出现失败情况，分析一下这 5 步出现异常后上游业务和消息发送是否一致：

上游应用执行完成，下游应用尚未执行或执行失败时，此事务即处于 BASE 理论的 Soft State 状态。

3.2 下游应用监听 MQ 消息并执行业务

1. 下游应用监听 MQ 消息并执行业务，并且将消息的消费结果通知可靠消息服务。(本地消息落地)

2. 可靠消息的状态需要和下游应用的业务执行保持一致，可靠消息状态不是已完成时，确保下游应用未执行，可靠消息状态是已完成时，确保下游应用已执行。 下游应用和可靠消息服务之间的交互图如下： 

   

3. 下游应用监听 MQ 消息组件并获取消息, 并存储本地消息

4. 下游系统通知可靠消息服务已接收到消息

5. 可靠消息把消息更新为已接收状态

6. 下游应用根据 MQ 消息体信息处理本地业务

7. 下游应用向 MQ 组件自动发送 ACK 确认消息被消费

8. 下游应用通知可靠消息系统消息被成功消费，可靠消息将该消息状态更改为以消费,任务表状态修改为已完成。

1. 下游应用监听 MQ 消息组件并获取消息, 并存储本地消息

2. 下游系统通知可靠消息服务已接收到消息

3. 可靠消息把消息更新为已接收状态

4. 下游应用根据 MQ 消息体信息处理本地业务

5. 下游应用向 MQ 组件自动发送 ACK 确认消息被消费

6. 下游应用通知可靠消息系统消息被成功消费，可靠消息将该消息状态更改为已消费,任务表状态修改为已完成

3.3 生产者消息状态确认

可靠消息服务定时监听消息的状态，如果存在状态为待确认并且超时的消息，则表示上游应用和可靠消息交互中的步骤 4 或者 5 出现异常。

可靠消息则携带消息体内的信息向上游应用发起请求查询该业务是否已执行。上游应用提供一个可查询接口供可靠消息追溯业务执行状态，如果业务执行成功则更改消息状态为已发送，否则删除此消息确保数据一致。具体流程如下:

3.4 消费者消息状态确认

下游消费MQ服务异步通知可靠消息的过程中可能出现异常,在此可能导致两个现象一、消息已接到但可靠消息没有确认接到二、消息已成功消费但可靠消息没有确认接到,为此下游系统需要提供消费者消息状态查询接口,从而可靠消息重新确认.在确认过程中如果是可靠消息为已消费而下游消费系统为已接收则不进行更新操作. 具体流程如下：

3.5 消息重投

消息已发送则表示上游应用已经执行，接下来则确保下游应用也能正常执行。 可靠消息服务发现可靠消息服务中存在消息状态为已发送并且超时的消息，则表示可靠消息服务和下游应用中存在异常的步骤，无论哪个步骤出现异常，可靠消息服务都将此消息重新投递到 MQ 组件中供下游应用监听。

 下游应用监听到此消息后，在保证幂等性的情况下重新执行业务并通知可靠消息服务此消息已经成功消费，最终确保上游应用、下游应用的数据最终一致性。具体流程如下：

1. 可靠消息服务定时查询状态为已发送并超时的消息

2. 可靠消息将消息重新投递到 MQ 组件中

3. 下游应用监听消息，在满足幂等性的条件下，重新执行业务。

4. 下游应用通知可靠消息服务该消息已经成功消费。

5. 更新consumer消息记录为已消费

3.6 删除上游系统7天前成功发送的消息

在预发送执行MQ消息的时候本地消息如果落库则需要删除消息,否则业务系统需要额外提供查询消息发送状态接口, 这里介绍两种方法

第一种,RPC服务接口来实现, 在生产者和消费者注册到可靠消息的时候把生产者和消费者存储到BeanFactory的Map里在定时清理任务的时候去处理在线的RPC服务

第二种,发可靠消息来实现, 确保100%到达

3.7 删除下游系统7天前成功消费的消息

在消费MQ消息的时候本地消息如果落库则需要删除消息,否则业务系统需要额外提供查询消息发送状态接口,删除实现同3.6

3.8 每天备份可靠消息记录

每天将成功消息删除并备份到对应数据库提供历史消息查询功能,当然如果你选择mongo可以不考虑备份消息

这里做一个说明，因为项目采用的是rocketmq，一个topic对应一个生产者，而可靠消息采用的是中间件负责发送消息，又不能采用中间件的生产者为所有上游系统发送消息，这里引入了zookeeper做注册中心，所以依赖可靠消息的服务，在启动项目的时候会像中间件去注册生产者，而中间件的watch机制会及时的更新生产者和消费者状态，而中间件会为使用中间件的系统提供sdk，使用者无需关注实现，只需要引入中间件的sdk和对应的注解即可完成可靠消息的发送和消费，详见下图： 普通消息发送流程： 

可靠消息发送流程：

可靠消息发送和消费流程： 

服务注册

消费注解 @MqProducerStore

生产注解@MqConsumerStore

定时清理所有订阅者消费成功的消息数据

定时清理所有生产者发送成功的消息数据

定时清理所有生产者发送成功的消息数据

处理发送中的消息数据

处理待确认的消息数据

可靠消息用法

例子

强制： 需要使用的使用加上述两个注解，方法参数需要加入 MqMessageData

如果对本文感兴趣，或者本文对您有所帮助，可靠参考github代码，本套代码是spring cloud E版本 + vue两套全家桶实现

如果有时间最好能给点加个星或者follow一下，笔者在这里先谢过了。对不知道怎么加星的朋友，请用电脑登录github或者码云,这里两个截图 

更多内容请参考paascloud 建站文档：

• https://document.paascloud.net/