基于ActiveMQ的消息中间件系统 OneMM逻辑与物理架构设计详解

1. 基本介绍与组件架构图

维基百科对消息中间件的定义是“ Message-oriented Middleware is software infrastructure focused on sending and receiving messages between distributed system。”分布式系统中实现消息发送和接受的基础设施。随着企业信息化建设的不断深入，多种业务应用相互关联，容易造成底层数据分散，应用系统间的耦合度高。针对该问题应从整体上调整目前系统架构，面向不同业务应用提供统一的数据访问服务，使用消息中间件对不同系统间的交互进行解耦，消息中间件技术有两个核心功能： 异步和解耦 。这两个核心功能整体上提高了应用系统的工作效率，增强了系统的可用性、稳定性和可扩展性，提升了用户体验。使用OneMM消息中间件系统可以实现应用系统各模块间或应用系统与其他系统（如ERP系统、支付系统）之间的解耦与异步消息传输，改变直接通过数据库共享方式交换数据，造成系统之间底层数据耦合度过高问题以及远程跨地域应用系统的数据交换问题。

OneMM系统组件逻辑架构图如下：

消息中间件架构主要组件说明：

（ 1）消息发送组件： 接收应用服务器发起的外系统调用请求，并将请求消息发送到本地消息队列或跨中心消息队列；

（ 2）消息接收组件： 接收本地消息队列或跨中心消息队列中的请求消息，并调用外系统提供的业务接口；

（ 3）消息转发组件： 消息转发器帮助跨中心应用进行消息转发，如图中的中心 1和中心2之间，应用可以通过消息转发器进行消息互发；

（ 4）消息代理组件： 消息代理组件主要负责接收跨中心消息，并将跨中心消息根据相关参数发送到下图中心 1的“请求处理结果Topic订阅消息队列“或分发到中心2的相关具体业务队列；

（ 5）队列管理组件： 配置管理不同中心下的全部消息队列，及相关调用接口；

（ 6）消息队列组件： 提供消息队列创建、销毁等队列操作及管理功能，架构使用 ActiveMQ开源消息队列工具。

（ 7）消息缓存组件： 本地或跨中心异步消息通过 Redis缓存组件获得返回结果消

2. 消息流转架构图

2.1 同步消息流转架构： 消息发送组件创建会话发出异步消息后，同步流程会话循环接收 Topic消息回复订阅队列的广播消息，通过会话唯一标识判断是否为该会话的结果消息，是的话返回给应用，否则忽略该消息，超时后直接返回超时。同步消息流转架构图如下：

2.2 异步消息流转架构： 本地或跨中心异步消息通过 Redis缓存组件获得返回结果消息。异步消息结果，通过新建会话返回。生产者发送消息后关闭会话，无需等待。完全的异步流程无需返回结果消息，用户通过发起查询请求通过缓存或本地系统数据库获得结果。但是在某些项目实际使用中，异步会话不会立即关闭而是等待结果，被调用者执行请求后将结果按照异步消息发送流程主动发送回去。会话通过轮询Redis结果缓存，根据Session唯一标识符判断是否存在该会话的返回结果，获得结果后将结果返回给请求发起者。异步的特点是被调用者主动获取消息，并主动通过其他队列发送结果消息，这时被调用者可以根据自己的状态和处理情况判断是否（及如何）获取消息，是否（及如何）执行操作并返回结果，被调用者可以自行决定该如何操作，比如可以批处理。因此对于并发处理能力较弱的应用（被调用者）应采用异步方式主动接收消息，并主动创建会话返回处理结果消息，并根据实际情况决定如何处理及返回结果（比如可采用批处理及返回的方式），之后请求发起者通过新建会话在本地缓存或本地数据库中获取返回结果，真正实现应用间的解耦。异步消息流转架构图如下：

3. 跨中心消息时序图

3.1跨中心异步消息处理时序图一

异步消息结果，通过新建会话返回。生产者发送消息后关闭会话，无需等待。完全的异步流程无需返回结果消息，用户通过发起新的查询请求，通过本地缓存或本地数据库获得结果。

3.2跨中心异步消息处理时序图二

在某些项目实际使用中，异步会话不会立即关闭而是等待结果，被调用者执行请求后将结果按照异步消息发送流程主动发送回去。会话通过轮询 Redis结果缓存，根据Session唯一标识符判断是否存在该会话的返回结果，获得结果后将结果返回给请求发起者。异步的特点是被调用者主动获取消息，并主动通过其他队列发送结果消息，这时被调用者可以根据自己的状态和处理情况判断是否（及如何）获取消息，是否（及如何）执行操作并返回结果，被调用者可以自行决定该如何操作，比如可以批处理。因此并发处理能力较弱的应用应采用该异步方式主动处理消息，并且请求发起者通过新建会话在本地缓存或本地数据库中获取结果，真正实现应用间的解耦。

3.3跨中心同步消息处理时序图一

会话发出异步消息后，会话阻塞，循环接收 Topic Response 队列的广播消息，通过Session 唯一标识符判断每次收到的Topic订阅消息是否为该会话的请求结果消息，是的话返回给应用，否则忽略该消息，继续循环接收广播，超时后直接返回超时。Topic消息并不会因为第一个消息没有被处理而发生阻塞。

完全的异步流程无需返回结果消息，用户通过发起查询，通过本地缓存或本地数据库获得结果；而跨中心同步异步化处理流程需要得到外系统处理结果。

3.4跨中心同步消息处理时序图二（使用临时队列）

同步消息结果，通过临时队列在一次会话中返回，临时队列中只保存本次会话中的消息。生产者需保持会话，等待结果返回。

4. 高可用（主从）与负载均衡架构图

消息发送中的接收 Topic订阅结果消息队列URL地址、消息接收队列URL地址、消息代理的发送与接收队列URL地址以及消息转发器发送的Topic结果消息队列URL地址，均需设置为Failover 地址。因为消息队列组件ActiveMQ是设置为主从的，因此任何组件连接消息队列的URL地址均需配置为主从Failover地址。

<broker xmlns="http://activemq.apache.org/schema/core" brokerName="localhost" dataDirectory="${activemq.data}" advisorySupport="false">

5. 消息队列组件 ActiveMQ配置

通过修改消息队列组件 ActiveMQ的配置文件activemq.xml，以“效率优先”的原则调整相关参数，配置说明如下：

5.1  ActiveMQ消息通知配置

消息通知实现了 ActiveMQ的Broker上各种操作的记录跟踪和通知。但是在使用临时队列实现同步消息时我们发现ActiveMQ产生了大量advisory通知消息，并且这些消息反复在网络中传输。这有可能与ActiveMQ 同步消息ACK机制有关。

取消消息通知的配置方法，在配置文件中增加如下配置：

<broker xmlns="http://activemq.apache.org/schema/core" brokerName="localhost" dataDirectory="${activemq.data}" advisorySupport="false">

5.2删除不活动的队列配置

ActiveMQ的 Queue在不使用之后，可以通过web控制台或是JMX方式来删除掉。也可以通过配置，使得Broker可以自动探测到无用的队列（一定时间内为空的队列）并删除掉，回收响应资源。由于ActiveMQ使用时自动创建Destination，并且默认情况下不会删除掉，这种只增加不减少，导致在queue创建频繁的情况下，本功能非常有用。

配置如下：

<broker xmlns="http://activemq.apache.org/schema/core" schedulePeriodForDestinationPurge="10000" advisorySupport="false">     <destinationPolicy>        <policyMap>           <policyEntries>              <policyEntry queue=">" gcInactiveDestinations="true" inactiveTimoutBeforeGC="10000"/>           </policyEntries>        </policyMap>     </destinationPolicy> </broker>

参数说明：

1） schedulePeriodForDestinationPurge： 10000  每十秒检查一次，

默认为 0，此功能关闭

2） gcInactiveDestinations：true  删除掉不活动队列，默认为false

3） inactiveTimoutBeforeGC：30000 不活动30秒后删除，默认为60秒

5.3死信队列配置

DLQ-死信队列 (Dead Letter Queue)用来保存处理失败或者过期的消息。出现以下情况时，消息会被再投递

1） A transacted session is used and rollback() is called.

2） A transacted session is closed before commit is called.

3） A session is using CLIENT_ACKNOWLEDGE and Session.recover() is called.

当一个消息被 redelivered超过maximumRedeliveries（缺省为6次）次数时，会给broker发送一个"Poison ack"，这个消息被认为是a poison pill，这时broker会将这个消息发送到DLQ，以便后续处理。缺省的死信队列是ActiveMQ.DLQ，如果没有特别指定，死信都会被发送到这个队列。缺省持久消息过期，会被送到DLQ，非持久消息不会送到DLQ

缺省所有队列的死信消息都被发送到同一个缺省死信队列 (ActiveMQ.DLQ)，不便于管理。可以通过individualDeadLetterStrategy或sharedDeadLetterStrategy策略来进行修改。

生产环境的配置如下：

<policyEntry queue=">">     <deadLetterStrategy>    <!-- Use the prefix 'DLQ.' for the destination name, and make      the DLQ a queue rather than a topic -->    <individualDeadLetterStrategy queuePrefix="DLQ." useQueueForQueueMessages="true" />     </deadLetterStrategy> </policyEntry>

queuePrefix:设置死信队列前缀

useQueueForQueueMessages: 设置使用队列保存死信，还可以设置 useQueueForTopicMessages，使用Topic来保存死信

5.4  消息队列数过多问题

默认的消息队列配置中使用一个独立的线程负责将消息存储中的消息提取到消息队列中，而后再被分发到对其感兴趣的消息消费者。如果有大量的消息队列 ,建议通过启用optimizeDispatch这个属性改善这个特性，示例代码如下所示：

<destinationPolicy> <policyMap> <policyEntries> <policyEntry queue=">" optimizedDispatch="true"/> </policyEntries> </policyMap> </destinationPolicy>

代码清单中使用通配符“ >”表示该配置会递归的应用到所有的消息队列中。

6. 总结

6.1为什么不能使用ActiveMQ临时队列实现同步消息

1）每一个消息生成一个临时队列，依然是并发访问没有产生“队列”的效果；

2）因为要通过临时队列维持每一个会话连接不断开，当大量会话同时到来时会造成网络拥堵，临时队列没有异步快速传递消息的效果；

3）同时也会造成一处堵塞，处处堵塞的火烧连营的效果。

6.2异步消息的优势和特点

所谓异步传输，就是当我将消息传递给你后，该消息就与我无关了，消息在不同应用或设备之间传递，但应用或设备彼此并不需要关联（在一次会话间建立长连接），即让消息通过不同会话在网络中进行短连接快速传递，实现应用或设备之间的解耦。异步传输的特征是会话层采用短连接，网络层更加稳定。

异步传输的关键在于：消息传递的速度。消息传输的速度越快，会话连接的时间也就越短，系统更加稳定（不会因为数据阻塞造成超时、网络阻塞、内存溢出等问题），系统的效率更高，就像人体的血液循环（提高血液循环的速度的关键在于，一是推动血液运动的“力”要足，二是血脂要低减少阻力）。因此可以说提高互联网应用稳定性的要诀在于：不断提高数据传输的速度。在大并发的情况下要做到这一点，一是要采用异步方式使用短连接传输，如果会话连接长时间不断开（长连接），当大量会话同时到来时会造成网络拥堵；二是对传输数据尽可能的压缩；三是提高数据处理的能力。

互联网开发无小事，因为互联网应用要面对海量信息，一个小小的问题也能被无限放大，就像汪洋中的蝴蝶效应。一个互联网应用（如本文所说的消息中间件）遇到的问题可能与系统、网络、应用、架构、数据等各个层面的问题有关，只有耐心发现并处理好每一个相关层面的问题，才能最终形成一款好的互联网应用。时刻提醒自己当下一波巨浪到来前我们会准备的更好。

作者简介： 王金剑， CSDN博客认证专家，开源数据库中间件Mycat核心开发成员。从事软件工作10年，曾在金融和互联网行业企业担任高级软件工程师、项目经理、高级DBA工程师职位，有银行信用卡系统和电子商务系统基础架构设计与开发经验。现在天狮集团担任高级DBA工程师，负责公司电子商务网站数据库设计与优化、数据库架构规划及部署规范制定、核心应用的软件设计与开发等工作。最近代表MYCAT团队参加第四届全国创新创业大赛，并在天津赛区决赛中胜出。