基于ActiveMQ的消息中间件利弊分析与实现

张振宇

摘要：消息中间件是利用可靠高效的消息递送机制帮助分布式系统进行平台数据交互的系统软件。分析了消息中间件的Java消息服务（JMS）标准、消息模式和发展现状，讨论了消息中间件带来的几个优势和应用消息中间件带来的若干问题，并提出了解决方案。以SpringBoot工程中应用ActiveMQ为例，通过编码实现展示了应用ActiveMQ的主要方法。

关键词：JMS；消息中间件；ActiveMQ

中图分类号：TP393.09文献标志码：A文章编号：1008-1739（2022）14-43-5

0引言

云计算是分布式计算的一种，主要是通过将庞大的应用服务拆分成多个较小的子程序，再分配给多台服务器进行处理，不同服务器的子程序之间必然需要通过网络进行通信协作，因而可靠、安全、高效的通信基础设施对于云计算的成功非常重要。

通常，消息通信机制分为同步通信和异步通信2种。在同步通信中，客户端直接请求服务端，并等待服务结果返回后才继续执行。这种实现方式结构简单，但是会造成客户端阻塞，导致程序服务并发访问效率低下，降低了服务的可用性。在异步通信中，客户端和服务端不会直接进行通信，而是客户端把请求以消息的形式放入消息队列中，服务端从消息队列中获取请求。这样客户端和服务端之间不会阻塞并等待对方响应，从而提高了并发访问的效率。

异步消息通信机制使云计算各层次中内部组件之间和各层次之间解耦合，同时有利于服务的动态伸缩性。异步通信机制经过多年发展，逐步形成了以JMS标准规范为代表的消息中间件体系。

1消息中间件

1.1 JMS

JMS是由SUN公司提出的基于Java平台消息中间件的接口规范，定义了消息的发送和接收等操作接口、消息格式和各个模块的功能语义[1]。JMS由生产者、消费者、服务提供者和消息等部分组成。JMS是在Java标准化组织内开发的标准，2001年6月发布了JMS1.0.2b版本，定义了点对点和发布订阅消息模式，当前已不建议使用。2002年3月发布了JMS1.1版本，引入了统一的接口定义来实现2种类型的消息传递。2013年4月发布了JMS2.0版本，在1.1版本基础上大大简化了接口操作，使用户能够更加专注于业务逻辑的开发工作[2]。

1.2消息模式

在JMS中定義了2种消息模式：点对点模式和发布订阅模式[3]。在点对点模式中，一条消息只能被一个接收者消费，是一对一的关系[4]，点对点模式如图1所示。

点对点模式有以下特点：

①一条消息仅能被一个接收者消费。

②消息并不是自动推送给接收者的，而是需要接收者主动从队列中请求获得。当有多个接收者监听同一个队列时，根据先请求先获得的原则确定接收者。

③如果没有接收者在监听该消息队列，消息会保存在该队列中，直到被消费或超时。

④接收者收到消息后必须向队列发送接收确认，否则消息中间件认为该消息没有被接收，仍可被其他接收者消费。

⑤发送者和接收者运行先后顺序没有限制。

在发布订阅模式中，一条消息可以被多个订阅者消费，是一对多的关系[4]。发布订阅模式如图2所示。

发布订阅模式有以下特点：

①消息中间件根据订阅者所订阅的主题将消息传递给订阅者。

②订阅同一个主题的多个订阅者会接收到同一条消息。

③消息会自动广播，订阅者无需主动请求或轮询主题获得新消息。

④订阅者必须先订阅主题，然后再等待发布者发布消息。

1.3消息中间件发展现状

目前主流消息中间件有ActiveMQ，RabbitMQ，ZeroMQ[5]。

ActiveMQ是一款由Apache组织发布的开源的（Apache License 2.0）独立的企业级通用消息中间件，完整地实现了JMS1.1和J2EE1.4规范，同时支持集群部署、事物控制、存储转发和持久化等一些企业级特性。ActiveMQ可以借助JVM跨平台的特性，可以运行于现阶段所有主流操作系统中。具备JMX和Web两种控制台，便于开发调试和运行监控，并且所有功能和特性都可以通过XML文件进行配置[6]。

RabbitMQ是一个开源的高级消息队列协议（Advanced Message Queuing Protocol，AMQP）实现，服务器端用Erlang语言编写，支持Python、.NET、Java，C语言、PHP等多种客户端；支持消息持久化和崩溃恢复，重新启动应用程序之后消息不会丢失；用于在分布式系统中存储转发消息，在易用性、扩展性和高可用性等方面表现不俗。

ZeroMQ是一个开源的、跨平台、高性能及精简灵活的网络消息中间件。支持新的轻量级socket风格的接口，支持多种底层协议和AMQP，支持多种平台和CPU架构。具有独特的非中间件的模式，不需要安装和运行一个消息服务器或中间件。只需要引用ZeroMQ程序库，就可以在应用程序之间发送消息，部署非常简单。性能方面比其他消息中间件要强。但是ZeroMQ不支持消息持久化和崩溃恢复，且稳定性较差。

2引入消息中间件的优势

2.1异步处理

在分布式服务情况下，一次请求可能会调用多个子系统的接口，需要等待所有的接口都处理完毕返回响应，才能获取最终的执行结果。这种同步接口调用的方式总耗时比较长，非常影响用户的体验，特别是在网络不稳定的情况下，极易出现接口调用超时问题。

例如，用户在注册界面提交注册信息后，系统需要处理注册信息并写入数据库，用户发送注册邮件通知注册成功，再向用户发送注册短信通知注册成功，都处理完后返回用户注册界面提示用户注册完成。假设3个业务节点处理时间都是50 ms，不考虑网络开销等其他处理时延，用户从提交注册到显示注册成功的等待时间是150 ms，同步处理流程如图3所示。

通过使用消息中间件将同步调用改为异步调用，将非必须的业务逻辑进行异步处理改造，能够显著减少系统响应时间。仍以用户注册为例，用户在注册界面提交注册信息后，系统需要处理用户注册信息并写入数据库，再写入消息队列，然后直接返回用户界面提示用户注册完成。向用户发送注册邮件通知和发送注册短信通知都改为异步处理。假设3个业务节点处理时间都是50 ms，写入消息队列耗时5 ms，不考虑网络开销等其他处理时延，用户注册等待时间是55ms，异步处理流程如图4所示。

2.2流量削峰

在类似秒杀的电商场景下，用户请求瞬时激增，所有的请求最终都压到数据库，可能会导致数据库响应变慢甚至无响应，传统模式流程如图5所示。

使用消息中间件之后，用户请求能够缓存在消息队列中，从而起到流量削峰的作用。订单系统接收到用户提交订单信息后，将请求直接发送到消息中间件，然后其他业务处理从消息中间件中获取订单信息进行处理和写库操作。由于消息队列服务处理消息速度比数据库快很多，如果出现请求峰值的情况，超量的消息能够暂存在消息中间件的队列中，订单处理系统会按照自己的处理能力来消费消息，不会对系统的稳定性造成影响，流量削峰流程如图6所示。

2.3系统解耦

复杂的业务系统一般都会拆分成多个子系统。以用户下单为例，请求会先通过订单系统，然后分别调用支付系统、库存系统、积分系统和物流系统，系统之间耦合性太高。如果调用的任何一个子系统出现异常，整个请求都会异常，对系统的稳定性非常不利。另外，如果需要增加商品推荐系统，也需要对订单系统进行修改，业务系统耦合关系如图7所示。

使用消息中间件相当于在各子系统之间插入了一个基于数据的隐含接口层，各子系统都要实现这一接口，从而在保证接口不变的前提下，允许各业务系统分别进行独立的修改。新增商品推荐系统时，也不需要修改其他业务系统，业务系统解耦后关系如图8所示。

3引入消息中间件的问题及解决方案

3.1可用性降低

引入消息中间件后，各业务系统都依赖消息队列服务，而一旦消息队列服务失效，会导致各业务系统之间无法通信，所以需要考虑消息服务故障导致的可用性降低问题。

针对单点故障导致的可用性降低问题，可以通過搭建消息服务集群来解决。消息中间件集群有多种实现方式，下面以ActiveMQ为例，通过Zookeeper搭建集群，消息服务集群如图9所示。

各ActiveMQ服务器会向Zookeeper注册，Zookeeper会为每个ActiveMQ服务器分配序列号，其中序列号最小的为主用服务器。生产者和消费者连接消息队列服务器时，会首先从Zookeeper获取当前主用服务器地址再进行后续访问操作。当主用服务器失效，Zookeeper会检测到并删除失效节点，然后选取当前最小序号的服务器升级为主用服务器，同时通知所有客户端，以完成主备切换，保证消息服务的高可用性。

3.2复杂性提高

系统引入消息中间件后，消息可能会被重复消费或者消息可能在消费前丢失。为了保证消息没有被重复消费和处理消息丢失的情况，系统复杂性不可避免地会有提高。

（1）消息重复问题

解决思路是增加一张消费信息表，给所有消息分配全局唯一的标识用作唯一索引。消费者开始消费之前，先去消费信息表中查询有没有匹配的消费记录，如果有就说明已经消费过，不再处理，消息重复解决流程如图10所示。

（2）消息丢失问题

解决思路是增加一张消息发送记录表，当生产者发完消息后，往该表写入一条数据，其状态标记为待确认。如果消费者读取消息后，调用生产者的接口更新该消息的状态为已确认。另外新增一个任务，定时检查消息发送表，如果一定时间间隔后，仍有待确认状态的消息，则认为该消息已经丢失，并重新发送该消息，消息丢失处理流程如图11所示。

3.3数据一致性

消息队列带来的异步可以提高系统响应速度，但是万一消息消费者在处理消息过程中没有正确处理，会导致数据不一致的情况出现。解决思路是通过在业务逻辑中增加重试机制来保持最终一致性。

对于消息量较小的业务场景，可以采用同步重试。在消费消息时如果处理失败，立刻重试若干次。如果还是失败，则写入到记录表中。如果消息量较大，不建议采用同步重试方式。如果出现网络异常，可能会导致大量的消息不断重试，影响消息读取速度，造成消息堆积。对于消息量较大的业务场景，可以采用异步重试。在消费者处理失败之后，立刻写入重试表，建立另外一个任务，定时读取重试表进行处理，数据一致性处理流程如图12所示。

4实验验证

本文以在Java主流开发的SpringBoot工程中引入ActiveMQ消息中间件为例，建立了一个消息生产者工程和一个消息消费者工程，同时在本机中运行5.7.0版本的ActiveMQ程序。

由上述代码示例可以发现，在SpringBoot工程中引入ActiveMQ消息中间件仅需要增加少量的代码。

5结束语

消息中间件为应用系统提供了高效、可靠的消息通信手段，能够很好地实现异构平台之间的信息交互。消息中间件以其独特的优势为各种分布式应用的开发注入了强大的活力，极大地推动了应用系统集成和发展。

参考文献

[1]陈安林.基于JMS的消息中间件的研究与实现[D].哈尔滨：哈尔滨工程大学，2006.

[2]朱方娥，曹宝香.基于JMS的消息队列中间件的研究与实现[J].计算机技术与发展，2008（5）：172-175.

[3]孫弋，温迅.一种面向消息的中间件的设计与实现[J].物联网技术，2019，9（3）：81-84.

[4] AHUJA S P， MUPPARAJU N.Performance Evaluation and Comparison of Distributed Messaging using Message Oriented Middleware[J].Computer and Information Science， 2014，7（4）：9-20.

[5]王小霞，陈亮.一种消息队列中间件的设计与实现[J].计算机工程，2006（21）：81-83.

[6] TIMOTHY B.Instant Apache ActiveMQ Messaging Application Development[M].Birmingham：Packt Publishing， 2013.