从JMS规范到ActiveMQ实战：消息中间件学习笔记

最近项目里有一段业务需要做异步通知，我顺手把Java消息服务（JMS）和Apache ActiveMQ完整过了一遍。之前工作中虽然用过现成的消息中间件，但坦白说对JMS规范本身的理解一直停留在“知道个大概”的层面。这次借着ActiveMQ的落地实践，从概念到代码把整条链路啃了一遍，发现很多以前模糊的地方清晰了不少。这篇算是整理出来的学习笔记，写给同样刚接触消息中间件、想系统搞懂JMS和ActiveMQ之间关系的朋友。你看标题里的“ActiveM”其实就是ActiveMQ，写的时候手滑少了个Q，懒得改了。

1. 整体学习思路：为什么要从JMS和ActiveMQ入手

1.1 消息中间件到底解决了什么问题

先把场景摆出来。我之前在一家做电商系统的公司，订单模块创建订单之后要通知库存系统减库存、积分系统加积分、推送服务发通知。最早这些逻辑全在订单接口里同步调用，服务之间耦合得很死，而且响应时间被下游拖得很难看。下单接口平均响应时间800多毫秒，大部分时间都耗在等下游返回结果上。后来把调用改成异步：订单创建成功后只发一条消息到消息服务器，下游系统各自订阅自己关心的消息去消费。接口响应时间直接降到了200毫秒左右，系统的吞吐能力也上来了。

这就是消息中间件解决的两个核心问题：异步解耦和流量削峰。放在更宏观的视角看，消息中间件是分布式系统里非常基础的一环，不管是订单系统、支付系统还是日志采集系统，都会用它来打通数据链路。

现在市面上消息中间件产品很多，像RabbitMQ、Apache Kafka、RocketMQ都各自有忠实用户。我第一次系统学习时没有直接上Kafka或RocketMQ，而是选了ActiveMQ，原因有几个：

• ActiveMQ是Apache开源项目，部署简单，不需要依赖额外的协调组件，下载解压就能启动，对新手非常友好。
• 它对JMS规范的支持最完整，是学习和理解JMS标准的最佳载体。
• 社区资料丰富，踩坑时能搜到大量现成的解决方案。
• 在中小规模系统、传统企业内部系统里，它的出场率其实比想象中高，学会它并不算浪费时间。

1.2 先搞懂JMS和ActiveMQ的关系

这是新手最容易绕晕的地方。简单说，JMS（Java Message Service）是一套规范，是Java世界里定义消息系统访问方式的统一API标准；ActiveMQ是实现这套规范的具体消息中间件产品。打个比方，JMS就像JDBC，ActiveMQ就像具体的数据库驱动，你通过JDBC接口操作数据库，底层连接的是MySQL还是PostgreSQL由驱动决定，而ActiveMQ就像是消息领域的“MySQL驱动”加上“数据库本身”。

这种关系决定了学习路径：先看JMS规范定义的API长什么样，再去理解ActiveMQ对规范的具体实现。规范定义了ConnectionFactory、Connection、Session、Destination、MessageProducer、MessageConsumer这些核心接口，ActiveMQ则提供了ActiveMQConnectionFactory等具体实现类。你写的业务代码只依赖javax.jms包下的接口，换一个JMS实现时，代码层面几乎不用改，只需要替换连接工厂的配置。这是JMS设计上最聪明的地方。

2. JMS核心概念梳理

2.1 消息模型：点对点与发布订阅

JMS规范定义了两种消息模型：点对点（P2P，Point-to-Point）和发布订阅（Pub/Sub，Publish/Subscribe）。这两个模型对应两个不同的Destination类型：Queue和Topic。

点对点模型使用Queue。生产者发送消息到Queue，消息服务器保存这条消息，然后由某一个消费者消费。关键特征是“一条消息只能被一个消费者消费一次”，多个消费者同时监听同一个Queue时，消息会在消费者之间竞争分发。消费完之后消息就从Queue中移除。这个模型天然适合任务分发场景：比如订单服务发一条“减库存”命令，库存服务只有一个消费者实例处理它，处理完就完事，不会重复处理。

发布订阅模型使用Topic。生产者发送消息到Topic，所有订阅了这个Topic的消费者都能收到这条消息的副本，就像广播。这个模型适合事件通知场景：比如用户下单后，库存服务、积分服务、短信服务都关心这个“订单创建”事件，它们各自订阅同一个Topic，各自独立消费。

从JMS API角度看，创建Queue和Topic的方式几乎一样：

Queue queue = session.createQueue("order.queue"); Topic topic = session.createTopic("order.topic");

2.2 核心API组件逐个说

JMS的API由几个核心组件构成，逻辑链路非常清晰。

ConnectionFactory是连接工厂，用来创建Connection。在ActiveMQ中，你创建的是ActiveMQConnectionFactory实例，构造参数传入broker的连接地址：

ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");

Connection代表应用程序到消息服务器之间的物理连接。创建Connection的开销比较大，包含了网络连接的建立和认证，实际项目中应该复用。Connection创建后需要调用start()方法启动连接，否则消费者收不到消息，这是个非常隐蔽的坑。

Session是JMS中最核心的对象，它提供了创建消息、生产者和消费者的方法。Session可以理解为一次“会话上下文”，它是单线程使用的。创建Session时可以指定两个参数：是否使用事务，以及消息确认模式：

Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

Destination是消息目的地，就是上面说的Queue或Topic。MessageProducer负责发送消息，MessageConsumer负责接收消息。Message是消息本身，是生产者与消费者之间传递的数据载体。

消息对象分三个部分：消息头（Header）、属性（Properties）和消息体（Body）。消息头包含了一些标准的元数据，比如JMSDestination、JMSDeliveryMode、JMSMessageID、JMSTimestamp、JMSCorrelationID等。属性部分可以设置自定义的键值对，这些属性除了携带业务数据之外，还能被用于消息选择器做过滤筛选。消息体是实际的数据内容，JMS规范定义了五种消息体类型：

日常业务中TextMessage用的最多，因为JSON格式通用性最好，跨系统通信时对方不一定是Java。

2.3 消息确认机制和事务：这块必须吃透

JMS可靠性的关键在于确认机制。消费者从broker收到消息后，需要在Session层面做“确认”，告诉broker这条消息我处理完了，你可以移除它了。如果消费者一直不确认，broker就会认为消费失败，消息会再次投递。

确认模式在创建Session时指定，一共有几种：

• AUTO_ACKNOWLEDGE：自动确认。消费者收到消息后，JMS实现自动帮我们确认，无需手动干预。这是最简单的模式，也是大多数入门示例使用的模式。
• CLIENT_ACKNOWLEDGE：客户端确认。消费者收到消息后必须手动调用message.acknowledge()方法才会确认。如果忘了调用，消息可能会被重复投递。
• DUPS_OK_ACKNOWLEDGE：允许重复确认。这种模式下确认是延迟的，消息可能被重复投递多次，可靠性和性能的折中方案。
• SESSION_TRANSACTED：事务模式。如果Session以事务方式创建，需要手动调用session.commit()提交事务，消息才会被确认；调用session.rollback()则回滚，消息重新进入投递队列。

事务模式和确认模式是两套机制，注意区分。创建Session时第一个boolean参数控制是否启用事务：

// 非事务、自动确认 Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); // 事务模式，第二个参数会被忽略 Session session = connection.createSession(true, Session.SESSION_TRANSACTED);

我在实际项目中踩过一个坑：用了CLIENT_ACKNOWLEDGE模式，但业务处理逻辑比较复杂，偶尔抛出异常后忘记调用acknowledge方法，导致消息被重新投递，消费端出现重复处理。后来规范了写法，把acknowledge调用放在finally块里，或者直接统一用事务模式，用commit和rollback来兜底。

ActiveMQ还扩展了一个INDIVIDUAL_ACKNOWLEDGE模式，可以单独确认某一条消息，不展开说了，知道有这个东西就行。

3. 环境搭建与第一个可运行的Demo

3.1 下载安装与验证启动

我使用的ActiveMQ版本是5.15系列，JDK是Java 8。先去Apache官网下载tar.gz或zip包，解压到一个目录下。bin目录里有启动脚本：

• Linux系统执行：./activemq start
• Windows系统执行：activemq.bat start

启动成功后，默认监听两个端口：61616是OpenWire协议的连接端口，客户端通过这个端口连接broker；8161是Jetty Web管理控制台的端口，浏览器访问http://localhost:8161/，默认用户名admin，密码admin，可以在控制台里看到队列、Topic、连接数等运行状态。

验证启动是否正常，有几个简便方法。一是用netstat查端口：

netstat -an | grep 61616

二是直接写一个最简的客户端程序去连一下，如果连接成功并且能发送消息，说明环境没问题。

3.2 引入依赖与生产者代码

Maven项目引入ActiveMQ客户端依赖：

<dependency> <groupId>org.apache.activemq</groupId> <artifactId>activemq-all</artifactId> <version>5.15.15</version> </dependency>

注意activemq-all这个依赖会打包ActiveMQ所有相关的类，包括broker端和客户端。如果只是写客户端程序，也可以用activemq-client这个依赖，体积更小，但直接使用activemq-all对初学者更省事。

下面写一个最简生产者：向队列order.queue发送一条文本消息，模拟“订单创建成功”事件。

import org.apache.activemq.ActiveMQConnectionFactory; import javax.jms.*; public class OrderProducer { public static void main(String[] args) throws Exception { // 1. 创建连接工厂 ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616"); // 2. 创建连接并启动 Connection connection = factory.createConnection(); connection.start(); // 3. 创建会话，非事务，自动确认 Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); // 4. 创建目的地 Queue queue = session.createQueue("order.queue"); // 5. 创建生产者并指定持久化模式 MessageProducer producer = session.createProducer(queue); producer.setDeliveryMode(DeliveryMode.PERSISTENT); // 6. 创建消息并发送 String orderJson = "{\"orderId\":\"ORDER-20250101-0001\",\"amount\":299.00}"; TextMessage message = session.createTextMessage(orderJson); producer.send(message); System.out.println("订单消息已发送: " + orderJson); // 7. 释放资源 producer.close(); session.close(); connection.close(); } }

代码里有一行容易被忽略但很重要的配置：producer.setDeliveryMode(DeliveryMode.PERSISTENT)。持久化消息会写入ActiveMQ的存储（默认是kahaDB），broker重启后消息不丢失。非持久化消息只存在于内存中，broker一重启就没了。对于订单这类核心业务数据，必须使用持久化模式。

3.3 消费者：同步和异步两种写法

消费端有两种主流写法：同步receive和异步监听。

同步写法使用consumer.receive()方法，这个方法会阻塞当前线程，直到收到消息或超时：

import org.apache.activemq.ActiveMQConnectionFactory; import javax.jms.*; public class OrderConsumerSync { public static void main(String[] args) throws Exception { ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616"); Connection connection = factory.createConnection(); connection.start(); Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); Queue queue = session.createQueue("order.queue"); MessageConsumer consumer = session.createConsumer(queue); // 阻塞等待消息，设置5秒超时 Message message = consumer.receive(5000); if (message instanceof TextMessage) { TextMessage textMessage = (TextMessage) message; System.out.println("【同步消费者】收到订单消息: " + textMessage.getText()); } else if (message == null) { System.out.println("【同步消费者】等待消息超时，没有收到消息"); } consumer.close(); session.close(); connection.close(); } }

同步receive适合那种“轮询拉取”的场景，比如定时任务从队列里取一批消息去处理。

异步写法是设置MessageListener监听器，消息到达时由JMS实现自动回调onMessage方法，代码不需要主动等待：

import org.apache.activemq.ActiveMQConnectionFactory; import javax.jms.*; public class OrderConsumerAsync { public static void main(String[] args) throws Exception { ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616"); Connection connection = factory.createConnection(); connection.start(); Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); Queue queue = session.createQueue("order.queue"); MessageConsumer consumer = session.createConsumer(queue); consumer.setMessageListener(new MessageListener() { @Override public void onMessage(Message message) { if (message instanceof TextMessage) { try { TextMessage textMessage = (TextMessage) message; System.out.println("【异步消费者】收到订单消息: " + message); // 在这里处理业务逻辑，比如减库存、加积分 } catch (JMSException e) { e.printStackTrace(); } } } }); // 异步监听是回调式的，main线程不能提前退出，做线程阻塞演示 System.out.println("【异步消费者】已启动，监听中..."); Thread.sleep(60000); consumer.close(); session.close(); connection.close(); } }

异步监听有一个注意点：MessageListener.onMessage方法在JMS连接的工作线程中执行，这个方法返回后才会调用下一轮通知回调。如果你的业务处理时间较长，会阻塞后续消息的接收。所以不要在onMessage里做耗时的阻塞操作，要拆到独立线程池去处理。

3.4 消息选择器：按条件过滤消息

JMS提供了一种按条件接收消息的机制：消息选择器（Message Selector）。创建消费者时可以传入一个表达式条件，只有满足条件的消息才会被投递给这个消费者：

MapMessage mapMessage = session.createMapMessage(); mapMessage.setString("type", "order"); mapMessage.setDouble("amount", 299.0); producer.send(mapMessage);

消费端筛选：

MessageConsumer consumer = session.createConsumer(queue, "type = 'order' AND amount > 100");

消息选择器用的是类SQL的语法，只能引用消息属性和消息头的标准字段，不能引用消息体里的内容。这个功能适合在同一个目的地里区分不同业务场景的消息。

4. 实操中踩过的坑和排查技巧

4.1 连接被拒绝：先查端口再查地址

新环境里最容易碰到的问题就是连接时抛出异常，核心报错通常是Connection refused。这个问题的排查顺序非常固定：

• 用netstat确认61616端口有没有在监听。
• 确认连接的地址对不对，如果broker不在本机，IP地址一定要写对。
• 如果用了防火墙，记得放行61616端口。

有一次我发现端口监听正常，但客户端就是连不上，折腾了很久才发现是配置里把tcp协议写成了http协议。这种低级错误，排查时一定要先看连接URL的格式。

4.2 消息发了但消费者收不到

这个问题的原因通常是连接没有start。Connection对象的start()方法必须在消费者开始接收消息之前调用，很多入门代码会忽略这一步，结果就是生产者正常发送、broker也收到了消息，但消费者端就是没反应，因为连接根本没启动。

排查思路是去管理控制台看队列的Enqueue和Dequeue计数。如果Enqueue在涨、Dequeue不变，说明消息进队列了但消费者没有消费；如果消费者代码已部署而且日志里没有异常，优先检查connection.start()有没有调用。

4.3 确认模式不匹配导致的重复消费

用了CLIENT_ACKNOWLEDGE模式但代码里没有调用acknowledge()，这是重复消费的典型原因。消息会被broker反复投递，直到消费者确认成功。重复消费在业务上会造成很大麻烦，比如重复减库存、重复发短信。

解决思路有三种：一是对生产者和消费者统一使用自动确认模式；二是事务Session，commit和rollback强制管理；三是业务幂等处理，通过唯一业务键去重。实际项目中建议至少做好第三点，因为分布式场景下，消息超时重投、网络抖动导致的重复其实是常态。

4.4 消息丢失：持久化与非持久化的区别

默认情况下，如果生产者在send之前没有设置持久化模式，消息是非持久化的，broker重启之后消息就丢了。很多人在测试阶段没注意这一点，到了生产环境才发现broker一重启，队列里的数据全部清空。

ActiveMQ的持久化机制默认使用kahaDB，一种基于文件的内存+磁盘混合存储，它的性能可以满足大多数中小系统的要求。如果需要更高可靠性，可以切换为JDBC持久化，把消息存储到数据库表里，但写入性能会下降，要谨慎评估。

对核心业务消息，务必显式设置deliveryMode为PERSISTENT。对实时性要求高、丢失影响小的数据，如实时日志，用非持久化消息性能更好。

4.5 管理控制台的几个重要指标

ActiveMQ管理控制台的queue页面有几个指标要看懂：

• Number Of Pending Messages：队列中等待被消费的消息数量。这个值如果持续增长，说明消费速度跟不上生产速度。
• Number Of Consumers：当前消费者的数量。如果为0，说明没有消费者在线，消息会一直积压。
• Messages Enqueued：累计进入队列的消息总数。
• Messages Dequeued：累计被消费掉的消息总数。

例行检查时重点看第一个指标，正常情况下它应该在一个低位水平波动。如果持续上涨，就要看消费者是不是挂了、消费速度是不是太慢，或者下游出问题导致消费者反复异常。

4.6 关于connection.start()和close()的小规矩

JMS资源释放顺序是有讲究的，关闭顺序必须是producer/consumer先关，再关session，最后关connection。如果反着来，有时会出现连接已经关闭但消息还在发送中的间歇性异常。

还有一个小经验：connection的创建开销很大，每次发送消息都新建连接再关闭，性能极差，正确做法是复用一个Connection和Session，把producer做成轻量复用对象。当然这只是入门阶段知道方向就好，真到了高并发阶段就该上连接池了。

5. 最后再分享一点个人的学习体会

这次系统学习JMS和ActiveMQ，最大的收获不是学会了一个具体产品的API，而是理解了消息系统的通用模型。后来再看RabbitMQ和Kafka的文档，发现很多概念是相通的，比如持久化、确认机制、消费者组、消息堆积，底层逻辑都一样。ActiveMQ里这些概念用最直白的方式呈现，作为入门教材不可替代。

我觉得生产上使用ActiveMQ要特别注意的就是确认机制和持久化这两个点，最好一开始就养成显式配置的好习惯。很多线上P0级别的消息问题，不是ActiveMQ本身出了故障，而是使用方没有理解这两套机制，在错误的地方用了默认值。

这篇笔记到这里基本就结束了，如果你也在入门消息中间件，希望这篇内容能帮你少踩一些我踩过的坑。