一、背景简介
在系统开发初期，很容易出现这样一种情况：不同业务线上开发人员，因为技术栈和版本时间的影响，在选型的时候会优先使用自己熟悉的，例如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，等待服务相对空闲的时候再去发送。
消息中间件作为系统间解耦的稳定支撑，在服务层面管理时，需要具备清晰的设计路线，以及流程关键节点的监控和记录，确保整个链路的稳定和容错。
四、源代码地址
GitEE·地址
https://gitee.com/cicadasmile
Wiki·地址
https://gitee.com/cicadasmile/butte-java-note/wikis
![]()
