从我自己的经历说起吧。在接手一个内部系统重构时，需要把耗时较长的报表生成任务从请求线程里摘出去，后端不用一直傻等，前端也不用一直转圈。当时组里讨论了好几个方案，最后敲定用 JMS + ActiveMQ 做异步消息队列。我啃了不少文档、踩了不少坑，也把很多容易绕晕的概念理顺了。这篇就把我学习和落地过程中的理解、代码和踩坑记录梳理成一则学习笔记，希望能帮到正在学 JMS 和 ActiveMQ 的朋友。
刚开始看 JMS 的时候，最大的感觉是术语多、抽象，什么 Session、Connection、Destination，不看实际代码很难串起来。后来我用快递系统做类比，很多概念一下就通了。消息中间件本质就是一个中间转运站：生产者是寄件人，消费者是收件人，Broker 就是分拣中心。这个类比后面还会贯穿全文。
1. JMS 是什么，它到底在解决什么问题
先讲清楚一个东西：JMS 全称 Java Message Service，它是一套规范，不是具体的实现。这套规范把消息生产、消息投递、消息消费这些操作全部抽象成统一的 Java API，这样你用 ActiveMQ、RabbitMQ 或者其他支持 JMS 的中间件，核心代码结构是差不多的。换个中间件，改改配置，业务代码不需要大改。
那它到底解决什么问题？我觉得可以分成三类场景来理解。
第一类是异步解耦。比如用户下单后，系统要发短信、发邮件、更新积分，这些操作如果全串在请求线程里，接口就会很慢。把超时的任务丢进消息队列，订单接口马上返回，后面的消费端慢慢处理，用户感知到的速度就快很多。
第二类是流量削峰。秒杀、抢券场景，瞬时流量特别大，数据库很容易被打垮。让所有请求先走进消息队列，后端消费端按自己能力慢慢处理，系统就不会被冲垮。
第三类是应用间通信。不同的服务、不同的语言写出来的模块，常常需要互相传数据。通过 JMS 中间件，A 服务把消息丢到队列里，B 服务拉走去处理，两者无需直接建立网络连接，降低了耦合。
我当时做的报表生成任务，就是典型的异步化和流量削峰：一次生成可能要跑几分钟，而且经常大量并发触发，直接同步做会拖垮应用服务器，所以丢进排队再逐个消费，是最合适的落地方案。
注意：JMS 是面向 Java 的一套规范。如果你的系统完全是异构的（比如 PHP、Go、Python 混用），可能更适合直接用 AMQP（比如 RabbitMQ）或者别的跨语言协议。JMS 的强项在于 Java 生态内，语言统一时用起来非常顺畅。
2. JMS 的核心术语和生活化理解
我第一次看 JMS 文档，ConnectionFactory、Connection、Session、MessageProducer、MessageConsumer 这串名词排在一起，直接看懵了。用快递类比以后，这几个概念就清晰多了。
- ConnectionFactory：好比快递公司的营业点，你要寄件，先找个营业点拿到联系渠道。在 JMS 里，它负责创建 Connection，一般启动时初始化一个全局实例就够了。
- Connection：好比你和快递公司之间建立的一条专属运输通道。它代表客户端与 Broker 之间的物理连接，通常一个应用只创建一两个 Connection。
- Session：好比一次寄件业务的会话窗口。它是创建消息、生产者和消费者的工厂，也是事务的边界。一个 Connection 可以创建多个 Session。
- Destination：就是消息要去的地址。Queue（队列）是点对点模式的地址，Topic（主题）是发布订阅模式的地址。类比快递，就是寄到哪里，比如“某市某街道某小区”。
- MessageProducer：消息生产者，把消息塞到 Destination 去。类比寄件人把包裹放进快递柜。
- MessageConsumer：消息消费者，从 Destination 里取消息。类比收件人从驿站取包裹。
- Message：消息本身，JMS 规定了若干种消息类型：TextMessage、BytesMessage、MapMessage、ObjectMessage、StreamMessage。日常用的最多的是 TextMessage，因为它跨语言、跨版本的兼容性最好。
还有两个很容易忽略但很重要的概念：消息确认（Acknowledgement）和事务（Transaction）。
消息确认解决的是“消费者收到消息后，Broker 怎么知道可以扔掉了”这个问题。Broker 把消息投递给消费者后，不会立刻删除，需要消费者确认“我处理好了”，才会真正移除。如果消费者收到消息后没来得及确认就挂了，消息还会被重新投递。
事务解决的是“一组消息要么全部成功，要么全部失败”的问题，跟数据库事务的思想一模一样。在 Session 里开启事务，发送多条消息后统一 commit，任何一条失败都可以 rollback，那这一批消息就全部撤销。
理解了这些概念，再回来看 API 代码，就很顺了。所有 JMS 客户端代码，套路都是一样的：创建 ConnectionFactory -> 创建 Connection -> 创建 Session -> 创建 Destination -> 创建 Producer/Consumer -> 发/收消息。
3. ActiveMQ 为什么适合做 JMS 的入门与实战落地点
再说说为什么选 ActiveMQ，而不是 Kafka 或者 RabbitMQ。Kafka 的强项是海量日志吞吐、流式处理，但它原本是分布式日志系统思路，消息确认、顺序性、延迟这些跟业务消息中间件的诉求不太一样。RabbitMQ 是另一套 AMQP 协议生态，不能直接原生使用 JMS API。ActiveMQ 是 JMS 规范的原生实现，开箱即用，学习曲线平滑，对中小规模内部系统完全够用。
ActiveMQ 的两个主要版本方向也提一下。ActiveMQ Classic（5.x）是传统版本，功能完整，成熟稳定，我项目里用的就是这套。ActiveMQ Artemis（2.x 以后）是升级版引擎，性能更强、高可用架构也更现代。如果你是全新项目，可以直接从 Artemis 入手；如果你维护老系统，大概率还是碰到 Classic。我的笔记以 Classic 为主，但核心 JMS 代码两者通用。
我的学习路径建议是：先在本地把 ActiveMQ 跑起来，用眼睛看一下管理控制台里队列和消息长什么样，再去写第一行代码。盲写代码但脑子里没有画面，很容易卡壳。
3.1 本地安装与快速启动
安装 ActiveMQ 5.x其实很简单：
# 下载解压后进入目录 cd apache-activemq-5.18.x # 启动（默认模式） bin/activemq start # 查看控制台日志 tail -f data/activemq.log # 管理控制台地址 # 默认 8161 端口，打开 http://localhost:8161管理控制台默认账号密码是 admin/admin，登录后可以看到队列、主题、消费者数量、未消费消息数等信息。我第一次练手时，故意不写消费者，往队列里发了几条消息，然后跑到控制台去看，队列里确实能看到消息堆积，那个画面让我对“Broker 中转”立刻有了直观认知。
如果启动报端口占用，通常是因为 61616（OpenWire 端口）或 8161（Web 控制台）被占用，修改 conf/jetty.xml 里的端口配置，或者直接换个空闲端口。
注意：ActiveMQ 5.x 默认监听 0.0.0.0，生产环境必须做访问控制和防火墙限制，不能把管理控制台和 OpenWire 端口裸奔到公网，这是个非常现实的安全点。
3.2 生产落地的两个关键配置文件
部署到服务器时，最常用的两个配置是 conf/activemq.xml 和 conf/jetty.xml。
activemq.xml 负责核心配置，包括持久化适配器、传输连接器、认证插件等。jetty.xml 则是 Web 管理控制台的配置。最常见的调整就是把默认的 KahaDB 存储路径改到有足够磁盘空间的目录，防止磁盘写满导致 Broker 崩溃：
<persistenceAdapter> <kahaDB directory="/data/activemq/kahadb" /> </persistenceAdapter>如果你要开启连接密码认证，则在 broker 标签下加一个简单认证插件，配置用户名和密码：
<plugins> <simpleAuthenticationPlugin> <users> <authenticationUser username="app-user" password="app-pass" groups="users"/> </users> </simpleAuthenticationPlugin> </plugins>这些配置不复杂，但每一条都可能直接影响线上稳定性。改完配置文件都需要重启 Broker 才生效，测试环境改完务必验证一遍。
4. 动手实战：Producer 和 Consumer 的关键代码与运行细节
本地环境准备好了，就可以开始写代码了。我先用 Maven 建一个普通 Java 工程，引入 ActiveMQ 客户端依赖。版本号要和 Broker 版本匹配，我用的版本组合是 ActiveMQ 5.18.x 对应客户端依赖 5.18.x，不要混用大版本，否则可能出现协议兼容问题。
<dependency> <groupId>org.apache.activemq</groupId> <artifactId>activemq-client</artifactId> <version>5.18.2</version> </dependency>4.1 生产端代码：发送一条消息到队列
先写一个最简生产者：
public class SimpleProducer { private static final String BROKER_URL = "tcp://localhost:61616"; private static final String QUEUE_NAME = "demo.queue"; public static void main(String[] args) throws Exception { // 1. 创建连接工厂 ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(BROKER_URL); // 2. 创建连接并启动 try (ActiveMQConnection connection = (ActiveMQConnection) factory.createConnection()) { connection.start(); // 3. 第一个参数 true 表示开启事务 Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE); // 4. 创建队列目标 Destination destination = session.createQueue(QUEUE_NAME); // 5. 创建生产者 MessageProducer producer = session.createProducer(destination); // 6. 创建文本消息 TextMessage textMessage = session.createTextMessage("hello jms + activemq"); // 7. 发送 producer.send(textMessage); // 8. 提交事务，消息才会真正进入队列 session.commit(); System.out.println("消息发送成功: " + textMessage.getText()); } } }这里有几个细节要注意，都是新手容易踩的坑。
第一个是事务模式下一定要记得 commit。很多人代码写完了，消息似乎也 send 成功了，去控制台看却什么都没有，原因往往是忘了 commit。没有提交的事务，消息不会真正被 Broker 接收。
第二个是连接工厂的 URL。默认tcp://localhost:61616是 ActiveMQ Classic 的 OpenWire 协议端口，不要用 61616 去连 8161，这是很常见的混淆。8161 是 Web 控制台，61616 才是消息端口。
第三个是 try-with-resources 的使用。Connection 和 Session 都实现了 Closeable，用 try-with-resources 能保证程序结束前正常释放资源，避免连接一直占着不释放。
提示：生产者的 Connection 启动不是必须的，但建议养成习惯统一
connection.start()，避免后续某些消费者场景下忘记启动而收不到消息。
4.2 消费端代码：同步拉取与异步监听
消费者有两个模型，一个是同步阻塞式consumer.receive()，一个是异步回调式setMessageListener。
同步接收代码结构如下：
public class SimpleConsumer { private static final String BROKER_URL = "tcp://localhost:61616"; private static final String QUEUE_NAME = "demo.queue"; public static void main(String[] args) throws Exception { ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(BROKER_URL); try (ActiveMQConnection connection = (ActiveMQConnection) factory.createConnection()) { connection.start(); Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); Destination destination = session.createQueue(QUEUE_NAME); MessageConsumer consumer = session.createConsumer(destination); // 同步阻塞，1秒超时 Message message = consumer.receive(1000); if (message instanceof TextMessage) { TextMessage textMessage = (TextMessage) message; System.out.println("收到消息: " + textMessage.getText()); } } } }这里我还是建议用receive(long timeout)而不是裸receive()，因为裸 receive 会一直阻塞，如果队列里没有消息，线程就永远挂在那儿了。加个超时设置，逻辑更可控，代码也更适合放在后台任务里。
异步监听是生产上更常用的方式。它的好处是消费者线程不用自己写 while 循环，Broker 有消息推过来，回调方法自动执行：
public class AsyncConsumer { public static void main(String[] args) throws Exception { ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616"); try (ActiveMQConnection connection = (ActiveMQConnection) factory.createConnection()) { connection.start(); Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); Destination destination = session.createQueue("demo.queue"); MessageConsumer consumer = session.createConsumer(destination); consumer.setMessageListener(message -> { try { if (message instanceof TextMessage) { System.out.println("异步收到消息: " + ((TextMessage) message).getText()); } } catch (Exception e) { // 这里要特别小心，异常必须被捕获，否则监听线程容易出问题 } }); // keep alive Thread.sleep(10000); } } }异步监听里最容易犯的一个错误，就是回调方法内不处理异常。因为 JMS 的消息监听回调是在 JMS 服务线程里跑的，如果回调抛出 RuntimeException，线程可能被中断，消息也可能被重新投递，造成“看起来消费者崩了”的问题。所以回调方法第一件事就是做好异常兜底。
注意：异步监听模式下的消费者比较特殊，代码里不要用
Thread.sleep来模拟长任务。如果回调业务处理时间长，可以另外开工作线程池处理，把消息转发给线程池之后立即返回。否则 JMS 回调线程被占满，后续消息推送会被堵住。
4.3 Queue 与 Topic 在代码上的差异和实际表现
很多人以为队列和主题就是换一个 API 的问题，实际上行为差异很大。
Queue 是点对点模型，一条消息只会被一个消费者消费。你启动多个消费者监听同一个队列，消息会分摊到不同消费者手里。这是天然的负载均衡场景，适合任务分发、订单处理这类业务。
Topic 是发布订阅模型，一条消息会被所有在线的订阅者各收到一份。每个订阅者相当于一个独立的消费者组。适合广播通知、实时行情推送这类场景。
在代码上，两者的差异其实就是一个 API：
// 队列 Destination destination = session.createQueue("demo.queue"); // 主题 Destination destination = session.createTopic("demo.topic");但 Topic 有个很关键的注意事项：普通消费者如果先启动并订阅了主题，生产者再发消息，能收到。如果消费者晚于生产者启动，或者中间断开重连，那就收不到那段时间内的消息了。因为普通订阅是“在线订阅”，离线消息不保留。
如果你想让后启动的消费者也能收到之前漏掉的消息，JMS 提供了持久订阅（Durable Subscription）机制。它要求消费者必须设置一个持久化客户端 ID，Broker 会为它保留离线期间的消息，等它重新上线后再推送。
connection.setClientID("client-1"); // 创建持久订阅 TopicSubscriber subscriber = session.createDurableSubscriber(topic, "subscriber-1");这里有个细节：持久订阅的名字和连接客户端的 ID 一旦固定，就不要轻易改，因为 Broker 靠这两个标识区分不同的订阅者。换了名称，旧的消息就找不回来了，就相当于换了一个入坑身份。
我自己的项目里，Topic 用得不多，大部分实际需求其实用 Queue 就够。做多模块广播时，会更倾向于让每个接收方自行创建队列并订阅，进一步降低耦合。
5. 消息可靠性背后的机制与实操抉择
对于业务系统来说，消息不能丢、不能重复是刚需。JMS 这套体系设计了三个层面来保证可靠性：消息确认（ACK）、事务、持久化。我把学习时的理解都说清楚。
5.1 ACK 确认机制，Broker 与消费者之间的握手
简单说一下 ACK 的四种常见模式：
- AUTO_ACKNOWLEDGE：自动确认。消费者顺利收到消息，没有抛异常，就自动回执告诉 Broker 可以删了。这是在代码里最常见的模式。
- CLIENT_ACKNOWLEDGE：客户端手动确认。消费者需要显式调用
message.acknowledge()，Broker 收到确认后才会删除消息。适合处理流程长、需要业务代码控制确认时机的场景。 - DUPS_OK_ACKNOWLEDGE：允许重复确认，性能最好，但可能丢消息。严格的业务系统一般不用。
- SESSION_TRANSACTED：配合事务使用，消息是否确认取决于事务是否提交。
我第一次看到 CLIENT_ACKNOWLEDGE 时有个疑问：手动确认的意义到底在哪里？后来在对接外部短信通道时明白了。消费端拉取消息到本地，调用外部接口发送短信，外部接口超时或返回错误，但如果代码里用的是 AUTO_ACKNOWLEDGE，消息在消费者“拿到”那一刻就已经被确认了，超时错误并不会触发重发，这条短信就悄悄丢了。
改成 CLIENT_ACKNOWLEDGE 后，我们可以先拿到消息，执行完整业务逻辑，确认外部通道成功后再手动 acknowledge。如果失败，就不确认，Broker 会重新投递这条消息，短信就不会丢。这是“至少一次投递”模式下的常见做法。
Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE); // 收到消息并成功处理业务后 message.acknowledge();注意：
acknowledge()是作用在当前 Session 内所有未确认消息上的，它会一次性确认这个 Session 里收到的所有消息，而不是只确认调用它的那条。如果你在同一 Session 里同时处理多条消息，不想一次全确认，那就得考虑按消息拆分 Session，或者用事务模式。
5.2 事务模式：消息的批量提交与回滚
事务在 JMS 中的作用是保证一批消息的原子性。用createSession(true, ...)开启事务模式下，第一个参数是是否使用事务，第二个 ACK 参数会被忽略。
事务模式最常见的场景是：消费端从队列批量拉取一批消息，比如 100 条，全部处理成功后commit()，只要有一条失败，就rollback()，这 100 条消息全部会被重新投递。
它的好处是保证批量一致性；坏处也很明显，回滚会造成大量消息重复投递浪费。所以事务粒度要把握好，一次事务最好只处理一个小批次，不要把几万条消息放进一个事务里。
生产端使用事务还有一层意义：批量发送消息时不开启事务，每条消息都要和 Broker 做一次网络请求，性能较差。开启事务后，多条 send 可以攒在一起，commit 时一次性发送，吞吐量会有明显提升。尤其是大量批量消息入库的时候，这个优化非常有效。
事务模式适合：批量审批通知、大批量数据同步 非事务自动确认模式适合：实时性要求高、单条消息处理快的场景5.3 持久化：Broker 挂了消息还能找回
ActiveMQ 的持久化解决的是“Broker 宕机后，消息是否还能恢复”的问题。发送消息时，可以在 producer 上设置发送模式来决定这条消息是否持久化：
producer.setDeliveryMode(DeliveryMode.PERSISTENT);如果设置成 PERSISTENT，消息会被写入磁盘存储。Broker 意外重启后，未消费的消息依然在。如果设置成 NON_PERSISTENT，消息只存在内存里，Broker 重启后消息就没了，但性能更好。
ActiveMQ 5.x 默认使用 KahaDB 作为持久化存储，它是一个基于日志文件的本地数据库，专门针对消息存储优化过。
有个生产经验值得记住：持久化和发送确认是两回事。即使消息已经持久化，发送端的send()方法也不过是告诉你在语义上消息已被收下，而不是说“已经入盘了”。如果你要的是权衡可靠性，可以在doPublish这类方法里再结合事务使用。
建议：生产环境一律开启 PERSISTENT，不要为了省那点性能去关持久化。等真的遇到 Broker 重启导致消息丢失时，再提性能就晚了。
6. 常见问题与排查技巧实录
学习和上线过程中的踩坑记录，我总结成一张速查表，再挑几个典型的详细说。
| 问题现象 | 可能原因 | 排查思路 |
|---|---|---|
| 生产者连接超时 | 端口写错、Broker 没启动、防火墙拦截 | telnet 测试 61616 端口，查看 activemq.log |
| 消息发送了但消费者收不到 | 生产端忘了 commit、订阅 Topic 错过消息、消费者没 start | 看控制台队列堆积，检查代码 Session 是否开启事务 |
| 消费者一直收消息但业务重复执行 | 消费回调异常、ACK 超时导致重投 | 检查回调异常捕获、检查 ACK 模式 |
| Broker 突然内存暴涨 | 有大量非持久化消息堆积、消费者卡住 | 控制台看待消费消息数量，检查消费者是否死掉 |
| 监听回调不触发了 | 回调线程占满、连接失败未重连 | 查看日志、检查连接工厂 failover 配置 |
下面展开几个比较值得讲的。
6.1 消费者回调里出现异常导致消息无限重投
这是我调试时碰到的一个典型问题。消费端代码在回调里调用了远程接口，远程接口抛了 RuntimeException，如果没有 try-catch 包住，JMS 客户端收到回调异常后，消息会进入重投逻辑，导致同一条消息不断被消费，日志刷屏，业务数据也可能被重复写入。
解决思路分两层：
第一层是回调方法内必须做异常兜底，把单个消息的失败隔离出来。
第二层是把消息的失败语义明确了。如果这是一条业务上永远无法处理成功的“死信消息”，一直重投没有任何意义，应当捕获后记录到错误表或死信队列，让后续人工介入。
6.2 消费者连接断了以后没有自动重连
默认的tcp://localhost:61616连接方式，网络故障或 Broker 重启后，不会自动恢复连接。生产环境要使用 failover 协议，让客户端能在连接断开后自动重连到 Broker：
String brokerURL = "failover:(tcp://localhost:61616)?initialReconnectDelay=1000&maxReconnectAttempts=20"; ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(brokerURL);failover 协议是 ActiveMQ 留给生产环境的“保命”配置，它能在底层连接断开时，按照参数配置重连。配合异步监听模式，重连后消费者会自动恢复订阅，业务代码几乎无感。
6.3 手动确认模式下忘记 acknowledge 导致消息永远收不到新消息
用 CLIENT_ACKNOWLEDGE 模式时，如果消费者拉取消息后代码抛异常，没有调用 acknowledge，Broker 就不会移除消息。但这里有个隐蔽问题：如果你每次连的是同一条消息，一直拉取失败，那后续新消息就进不来，因为前面的消息堵住了队列头。
换一批新消息之前，要确保每一条都成功确认；或者用 try-finally 结构，确保在处理异常时也做出明确决策：是确认丢弃，还是回滚重投。不能模棱两可。
7. 学习路径建议与后续扩展方向
如果你刚开始学 JMS 和 ActiveMQ，我的建议路径是这样：
先把官方 Quick Start 跑一遍，然后自己写一个生产者、两个消费者同时监听队列，观察消息如何分摊。再写一个 Topic 的发布订阅 demo，对比两种模型的行为差异。接着把 ACK 模式、事务模式、持久化三个可靠性维度分别验证一遍，基本概念就牢固了。
后续扩展方面，有几个方向值得继续深入：
第一个是 Spring 集成。Spring 提供了 JmsTemplate 和@JmsListener注解，封装了大部分模板代码。用 Spring Boot 做集成时，需要配置 ConnectionFactory、JmsListenerContainerFactory，然后一个注解就能监听队列消息，代码非常简洁。但从学习角度，我强烈建议先写好原生 JMS API 再进 Spring，因为你理解了底层，用框架时才能明白它替你做了哪些事，遇到问题才知道去哪里排查。
第二个是失效转移和集群。ActiveMQ 可以配置共享存储下的主从模式，或者网络连接器（Network Bridge）做多个 Broker 互联，形成集群。集群引入后要考虑的不仅是消息怎么发送，还有消费者负载的均匀分配、订阅恢复的机制。
第三个是吞吐量压测。在把系统放上线之前，最好对队列做一轮简单的压测，把生产端的发送速度、消费端的处理速度、消息大小都记录下来，然后再决定线程池大小、事务批次大小这些参数。
我做完这个项目后最大的感触是：JMS 和 ActiveMQ 本身并不难，难的是想清楚每一层机制在解决什么问题。ACK 是为了“消费结果可感知”，事务是为了“多消息操作可原子”，持久化是为了“Broker 挂了数据不丢”。把这些机制放到真实业务里去理解，代码怎么写都顺手。
