初衷

为什么要设计消息总线（对消息队列进行二次封装），而不是让各业务系统直接使用RabbitMQ、Kafka、RocketMQ这样的成熟的消息队列呢？
如果业务系统比较简单，确实不需要考虑这样的问题，直接拿最成熟的开源方案是最好的方式，但是在复杂的多系统下、多人分工合作的场景下，直接使用成熟的消息队列一般都会面临以下问题

1. 开发难度大，各系统间分别隔离，需要关注消息中间件的各种复杂繁琐的配置，关注不同的消息则需要对接不同的消息队列
2. 维护成本高，各系统或团队需要分别管理消息中间件、处理各种服务异常、（消息中间件的高可用、业务的高可用等）
3. 管理难度大，没法对消息的生产和消费进行业务管理，也不方便对消息中的敏感数据进行权限管理
4. 扩展成本高，无法统一消息系统扩展功能，如路由、延时、重试、消费确认等
   总结消息队列是一个面向技术的接入，重点关注消息队列的配置、接口对接；而消息总线则是通过屏蔽部署、分组和通信等技术细节，实现一个面向业务的接入，重点关注要接收什么消息。

架构设计

消息总线屏蔽了消息发送、路由、分组、存储、消费负载、通信、高可用等一系列技术问题，对业务开发者来说只需要调用生产者或消费者SDK即可，简化了接入流程并提升了可靠性。
消息总线整体架构图如下图所示

基于RabbitMQ的消息总线架构

• 生产者通过对接消息总线提供的SDK进行Publish消息
• SDK充当消息队列Broker的角色，提供了消息路由、分组的功能、同时保证消息的高可用
• 本设计以RabbitMQ作为消息引擎，在具体应用中也可以使用其他引擎如Kafka，具体路由策略可在SDK中配置
• 消息的订阅关系可以存储在DB或者配置系统中
• 消费者应用系统并没有直接对接RabbitMQ，而是使用Deliver服务集群来负责把AMQP协议转换为HTTP或RPC协议，然后将消息传递到相应业务系统，该方式可以支持多种架构如单体应用、SOA、微服务等

消息体

为保证消息在整个流程中减少复杂度，将消息体设计为如下统一结构

消息体设计

• Header，通用数据，包含message ID、source、timestamp等参数
• Payload，具体消息体，设计为json格式，便于阅读以及在跨系统、跨语言之间传输
• Receiver，消息接收者，从订阅关系中获取，直接透传以提高效率

高可用

消息生产高可用

由于作为一个总线应用，会对接各种结构、语言、复杂度的系统，各系统的可靠性及处理能力差异较大，因此最直接想到的问题是各种类型的消息是否会相互干扰？

通过前面的介绍，可以了解到生产者发送的消息在经过Broker（SKD）时进行了路由、分组，具体策略如下：

1. 多分组消息间：消息总线可根据业务、流量等维度进行物理隔离，保障各消息不会相互干扰

2. 单分组消息内：在一个分组内消息总线可通过如下控制策略保障互不影响，具体设计如下

   
   分组内消息保护策略

消息经过Broker默认会进入一个RealTime Queue的队列中，Deliver集群中会有多个Listener监听RealTime Queue，在Deliver服务内，通过Dispatcher来控制消息并发及总数，当某种消息的发送量超过阈值时，就会触发处罚策略，将消息转发到Async Queue中，避免消费Worker都被同一个类型的消息占用，而Async Queue会被独立的Deliver服务监听进行消费，不影响RealTime Queue的消费，如果投递失败则会进入Retry Queue，满足一定条件后将会继续投递

消息消费高可用

为了保证消费时的高可用，Deliever集群在负责进行消费协议转换之外，也做了一些策略来保证消费端的高可用

• 熔断，在消息一段时间内失败数据超过阈值时，停止对队列的消费，避免由于服务抖动或者线上故障引起的大面积消息消费失败
• 自恢复，熔断后Deliver服务会对应用服务健康度进行监控，在服务恢复后可自动恢复消费
• 自失败重试，消息总线服务发生故障时，可对期间的失败消息采用重试策略进行重试，避免由于基础服务问题造成的消费失败
• 业务失败重试，在业务应用消费产生异常时，可在订阅消息时指定是否进行重试，消息总线会对需要失败重试的消息按照退避算法根据一定的时间间隔进行多次重试。
• 优雅重启，Deliver可实现优雅重启和退出，保障当前正在消费的消息都处理完成后才退出进程。