一、背景简介

在系统开发初期，很容易出现这样一种情况：不同业务线上开发人员，因为技术栈和版本时间的影响，在选型的时候会优先使用自己熟悉的，例如 MQ 中间件常用的：Kafka、Rocket、Rabbit 等，这样很容易忽略各个项目之间的组件差异问题；

在系统开发中后期，业务相对稳定之后，通常都会对资源占用较高的模块逐步重构，公共服务进行整合管理，从而使系统更具有整体性，在这个过程中，解决不同项目的中间件差异通常首当其冲，例如常见的缓存中心，MQ 消息管理等；

这种情况一般来说很难避免，系统初期为了快速支撑业务，埋下很多坑点，一旦业务可以稳定发展，并且可持续性得到验证，就会开始适当考虑逐步进行模块重构，降低成本。

二、重构思路

2.1 初期问题

在某创业公司研发初期，业务线上存在五个项目并行开发的情况，当时对于 MQ 的使用状况如下：

• Rocket：核心业务 3 个项目，版本有差异；

• Kafka：数据权重偏高，1 个项目采用；

• Redis：基于 Python 连接，队列消息模式；

刚开始因为用的不多，整体还在可控范围内，后续随着业务的持续迭代，项目间出现需要通信的情况，就开始混乱难以维护，然后就是被迫开始重构，统一消息组件。

2.2 二次选型

基于业务的综合考量，对现有几个项目进行 MQ 重新设计，形成的整体架构思路如下：

• MQ 组件选择：采用 RocketMQ；

• 换掉 Redis 组件的队列模式；

• 将基于 Python 的系统改 Java 语言；

• 提供消息生产与消费两个服务；

• MQ 的功能由上述服务进行统一维护；

这里在核心业务线上没有改变组件选择，换掉 kafka 的一个原因是涉及大量结算业务，Redis 队列模式弃用，基于 Python 的管理系统功能不多，这里只是顺手换掉，统一业务线的编程语言。这样设计之后，从整体思路上看就会合理很多。

三、改造过程

3.1 整体思路

涉及核心角色说明，从左向右依次：

• 生产客户端：需要请求服务端通信的节点，调用生产服务端封装的消息发送接口即可；

• 生产服务端：封装消息发送 API，并维护路由管理，权限识别等，消息落地存储等；

• 消息存储层：主要基于消息中间件进行存储，数据库层面用来处理特定情况下的二次调度；

• 消费服务端：封装消息接收 API，并根据路由标识，请求指定的消费端接口，完成通信；

• 消费客户端：响应消费服务端的请求，封装消费时具体的业务逻辑；

在整体的技术难度上没有太多差别，但是引入两个服务【生产和消费】，用来管理 MQ 通信流程，适配特定的业务逻辑，引入数据库做一次落地存储，在异常流程的处理上更加灵活，这样整个消息模块具有很强的可扩展性。

3.2 细节描述

• 组件选型

消息中间件的选择是比较多的，但是鉴于业务线上开发人员的熟悉程度，以及参考多方提供的测试对比报告，最终确定选用 RocketMQ 组件，同时 RocketMQ 相关特点：高性能、高可靠性，以及对分布式事务的支持，也是核心的考虑因素。

• 微服务架构

基于当前微服务的架构模式，把 MQ 功能本身集成在两个核心服务中，进行统一管理和迭代，以及组件的版本控制，对于所有生产的消息，进行全局路由控制，以及特定情况下的，通过应用服务层面功能设计，实现消息延时消费，以及失败消息的二次调度执行，和部分单条消息的手动触发。

• 数据存储

对消息实体进行二次存储，主要还是适配部分特定的功能点，有些消息可以延时处理，例如当 MQ 队列出现堆积的时候，或者达到监控的预警线时，可以通过配置手段，干预一部分消息只存储入库，不推送 MQ，等待服务相对空闲的时候再去发送。

消息中间件作为系统间解耦的稳定支撑，在服务层面管理时，需要具备清晰的设计路线，以及流程关键节点的监控和记录，确保整个链路的稳定和容错。

同系列：分布式概念 | 分布式事务 | Kafka集群 | RocketMQ组件 | Redis集群

四、源代码地址

GitEE·地址https://gitee.com/cicadasmileWiki·地址https://gitee.com/cicadasmile/butte-java-note/wikis

来源：https://www.cnblogs.com/cicada-smile/p/15322598.html