如果你是个写 Java 后端的早晚会撞上这几个字母JMS。我以前一直停留在“知道它是个消息中间件标准”的层面直到前几天决定动手把ActiveMQ拉起来完整跑一遍才把生产者、消费者、队列、主题这些概念一点一点拼完整。这篇就当是我的学习复盘从 JMS 设计思路开始讲到本地部署一个 ActiveMQ broker再到手写一份能跑通的生产消费代码最后把踩过的坑列成清单。适合刚接触消息队列、或者一直没亲手连过 ActiveMQ 的 Java 开发者。你要是想搞懂消息异步、解耦这些词到底对应代码里的哪些对象这篇应该能帮上忙。1. 先想清楚JMS 这门课到底在讲什么1.1 JMS 和消息中间件是什么关系先把最容易混淆的概念理清。**JMSJava Message Service**本身不是一个具体的软件而是一套 Java 标准接口。消息中间件是那个实际处理消息的程序ActiveMQ 就是这样的消息中间件。两者之间的关系有点像 JDBC 和数据库JDBC 定义接口数据库提供实现。你用 JMS 接口写的代码理论上可以对接任何实现了 JMS 规范的消息服务这层抽象是学习 JMS 最大的价值。比如常见的注册场景用户提交注册后如果直接同步发注册邮件、发短信通知万一某个服务响应慢用户等的时间就变长。把通知类任务放进消息队列用户接口只需要完成核心注册逻辑后把消息发出去后面谁消费这条消息、怎么处理都不再占用请求链路。消息中间件在这里起到的作用就是异步、解耦和削峰。为什么需要统一标准因为消息中间件有很多种如果没有 JMS 这样的接口层每个中间件都要用各自的 API业务代码就被绑死在一个具体产品上。JMS 抽象了 ConnectionFactory、Connection、Session、Destination、Producer、Consumer、Message 这些概念所以你不光是在学 ActiveMQ更是在学一套通用的消息编程模型。弄懂这套模型以后切换到别的消息服务成本会低很多。1.2 JMS 编程模型中的七个核心对象第一次看 JMS 文档的人容易被一堆接口名吓到其实整套模型可以理解成一条流水线ConnectionFactory负责创建 Connection是整个模型的入口。Connection代表客户端与消息服务之间的连接可以理解成一条网络链路。Session会话消息收发都靠它来组织生产者和消费者都在会话里创建。Destination消息发往哪、从哪收队列 Queue 和主题 Topic 都算它。MessageProducer负责发送消息。MessageConsumer负责接收消息。Message真正传输的内容。流程可以概括成一条链路从 ConnectionFactory 创建 Connection从 Connection 创建 Session再从 Session 创建 Destination 和 Producer最后由 Producer 把 Message 发到 Destination。Consumer 这边的步骤几乎一样只是把 Producer 换成 Consumer再调用 receive 或注册监听器来接收消息。Message 本身也有多种类型TextMessage、MapMessage、BytesMessage、StreamMessage、ObjectMessage。最常用的是 TextMessage适合普通文本和 JSON 字符串MapMessage 适合键值对数据。ObjectMessage 虽然能直接传 Java 对象但牵扯到反序列化安全问题实践里尽量少用。我一开始也图省事直接用 ObjectMessage后来了解到风险后就改成把对象序列化成 JSON 再放 TextMessage 里稳妥很多。1.3 队列模型与主题模型该怎么选JMS 规范定义了两种消息模型。第一种是点对点模型对应队列Queue。消息被发送到队列后只会被一个消费者取走这个消费者处理完并确认消息就从队列移除。适合任务分发、订单处理这种“每条消息只需要处理一次”的场景。第二种是发布订阅模型对应主题Topic。生产者把消息发到主题所有在线的订阅者都会各自收到一份一个订阅者收到不会影响另一个订阅者。适合事件广播、通知推送这种“一份消息要让多个接收方知道”的场景。有一个非常容易踩的选择误区想要“广播”效果却用了队列导致只有一个消费者收到想要“任务分发”效果却用了主题导致每个消费者都收到同一份数据。我在项目里见过类似问题一个服务开了多少个实例Topic 里的消息就被重复处理多少遍。选型的时候先问自己这个数据是要被多个独立系统消费还是只是同一套逻辑的不同实例轮流干活前者用 Topic后者用 Queue。对比点Queue 队列Topic 主题消息接收一条消息只能被一个消费者消费一条消息会广播给所有订阅者消息去向消费确认后从队列删除每个订阅者各自维护一份离线补收消息一直等谁有空谁取非持久订阅在线期间才能收到典型场景任务队列、订单异步处理事件广播、消息通知2. 为什么选 ActiveMQ 作为入门对象2.1 同类消息中间件的一点点对比想学消息中间件可选的产品不少。我选 ActiveMQ 不是因为它最流行而是因为它最适合用来理解 JMS 本身。ActiveMQ 是纯 Java 实现对 JMS 的支持比较完整安装方式就是解压一个包然后启动几乎没有额外环境依赖。你用 JMS 标准接口写出的代码在它上面跑是标准的学到的东西也最贴合规范本身。如果一上来就学 RabbitMQ虽然性能确实不错但它的核心协议是 AMQPJava 开发者接触到的接口跟 JMS 不是一套如果学 Kafka优势在分布式日志流和高吞吐很多概念是为大规模场景设计的入门门槛不低。不是说它们不好而是作为理解 JMS 的教学载体ActiveMQ 最轻、直接。另外要注意版本问题。ActiveMQ 5.x 系列是经典的 JMS 1.1 实现网上大量教程用的都是javax.jms包后来社区的重心放在 ActiveMQ Artemis 上接口包名换成了jakarta.jms。看教程的时候先分清楚项目用的是哪个版本不要把两套 API 混着抄否则编译能过都算运气好。2.2 本地启动配置端口、目录、日志下载 ActiveMQ 安装包并解压后你会看到一个很标准的目录结构。bin里是启动脚本Unix 系统用activemqWindows 用activemq.bat。conf里最重要的文件是activemq.xmlbroker 的传输端口、持久化方式、内存限制都在这里配置jetty.xml控制管理控制台user.properties存控制台的账号密码。data目录是运行时数据目录日志文件activemq.log也在里面。默认配置下只需要关心两个端口TCP 61616 是 OpenWire 协议端口Java 客户端连的是这个8161 是管理控制台的 Web 端口。看到 8161 不要以为是用来连接客户端的地址。我第一次排查问题看到 8161 能访问就想拿它当连接地址折腾了好一阵才反应过来其实客户端要连的是 61616。启动之前确认本机有 Java 环境ActiveMQ 5.x 至少需要 Java 8。然后在 bin 目录执行启动脚本。启动日志和错误都会写到data/activemq.log遇到问题先翻这个文件比瞎猜强得多。2.3 三步验证 Broker 已经可以干活我习惯用三步验证一个本地 broker 是否真的可用。第一步启动进程后等待几秒然后检查 61616 端口是否处于监听状态。Unix 环境可以用netstat -ano | grep 61616看到 LISTEN 再继续。第二步打开浏览器访问http://127.0.0.1:8161/admin默认账号admin、密码admin。控制台首页能看到 broker 名称、运行时间、连接数、消息数量等统计信息。第三步在控制台的 Queues 页面创建测试队列用页面的 Send 功能直接发一条消息然后去 Topics 页面再确认一下。这一步能快速验证 broker 的收发链路是通的。等我们自己写的消费者起来后还能用它配合验证。三步走完说明 broker 已经可以干活了。3. 上手写第一段 JMS 生产者与消费者代码3.1 工程依赖和连接信息的准备笔者的实验工程是标准 Maven 项目Java 版本 8 以上。依赖只需要一个dependency groupIdorg.apache.activemq/groupId artifactIdactivemq-all/artifactId version5.18.3/version /dependencyactivemq-all会把完整的 ActiveMQ 客户端和 JMS API 都带进来学习阶段不用纠结缺失的类。生产环境可以根据实际用到的特性换更精简的依赖比如activemq-client但初期没必要。如果项目日志框架比较敏感注意activemq-all可能跟 log4j 版本打架动手前看一眼依赖树。连接信息就一个地址tcp://127.0.0.1:61616。如果 broker 在另一台机器把地址换成对应 IP。生产环境可能会用failover:(tcp://host1:61616,tcp://host2:61616)这样的重连 URL入门阶段先用单节点最简地址就行。3.2 生产者实现从创建工厂到发送消息先写生产者。代码量其实不大核心就是创建工厂、建连接、开会话、建队列、发消息。关键对象都加了注释对照上一节说的七个核心对象会非常直观。import javax.jms.*; import org.apache.activemq.ActiveMQConnectionFactory; public class JmsProducer { public static void main(String[] args) throws JMSException { String brokerUrl tcp://127.0.0.1:61616; ConnectionFactory factory new ActiveMQConnectionFactory(brokerUrl); try (Connection connection factory.createConnection()) { connection.start(); try (Session session connection.createSession(false, Session.AUTO_ACKNOWLEDGE)) { Queue queue session.createQueue(queue.demo); MessageProducer producer session.createProducer(queue); for (int i 1; i 5; i) { TextMessage message session.createTextMessage(hello- i); producer.send(message); System.out.println(Sent: message.getText()); } } } } }解释几个小细节。第一Connection 创建出来后最好都调用connection.start()。虽然生产者不调用 start 有时也能把消息发出去但消费者那边如果不 start是绝对收不到消息的。统一都调用避免出现“生产者正常、消费者静默”的奇怪现象。第二createSession的第一个参数是是否启用事务第二个是消息确认方式。学习阶段先用非事务加自动确认的组合后面的可靠性部分会细讲。第三队列名queue.demo是我随便取的ActiveMQ 会自动创建不存在的队列所以不用去控制台提前建。这点对快速验证很友好。编译运行之后去控制台的 Queues 页面看queue.demo能发现消息数量增加了运行一次加 5 条很好验证。3.3 消费者实现同步接收和消息监听消费者有两种写法。第一种是用receive()同步接收。程序会阻塞在 receive 上有消息就返回可以传一个超时时间避免无限等下去。适合写测试脚本或者后台简单轮询。import javax.jms.*; import org.apache.activemq.ActiveMQConnectionFactory; public class JmsConsumerSync { public static void main(String[] args) throws JMSException { ConnectionFactory factory new ActiveMQConnectionFactory(tcp://127.0.0.1:61616); try (Connection connection factory.createConnection()) { connection.start(); try (Session session connection.createSession(false, Session.AUTO_ACKNOWLEDGE)) { Queue queue session.createQueue(queue.demo); MessageConsumer consumer session.createConsumer(queue); for (int i 1; i 3; i) { TextMessage message (TextMessage) consumer.receive(5000); if (message null) { System.out.println(等待超时队列里暂时没有更多消息); break; } System.out.println(Received: message.getText()); } } } } }第二种是用MessageListener异步监听。注册监听器之后消息到达时会自动回调处理程序可以继续做其他事情。适合真实业务里的消息处理。import javax.jms.*; import org.apache.activemq.ActiveMQConnectionFactory; public class JmsConsumerAsync { public static void main(String[] args) throws Exception { ConnectionFactory factory new ActiveMQConnectionFactory(tcp://127.0.0.1:61616); try (Connection connection factory.createConnection()) { connection.start(); Session session connection.createSession(false, Session.AUTO_ACKNOWLEDGE); Queue queue session.createQueue(queue.demo); MessageConsumer consumer session.createConsumer(queue); consumer.setMessageListener(message - { if (message instanceof TextMessage) { try { System.out.println(Async Received: ((TextMessage) message).getText()); } catch (JMSException e) { e.printStackTrace(); } } }); System.out.println(进入监听状态按 CtrlC 退出); Thread.sleep(30000); session.close(); } } }异步监听有两个坑。第一回调方法默认运行在 JMS provider 提供的线程里不要在回调里做特别耗时的操作否则会影响后续消息的处理节奏。第二lambda 里直接抛受检异常是编译不通过的所以getText()外面要 try-catch。看起来小真写代码时会卡一下。提示如果异步回调里要跑重逻辑尽量把任务交给业务线程池而不是在消息回调线程里直接处理避免阻塞后续消息消费。为了验证队列的点对点特性可以开两个消费者同时跑再发几条消息会发现每条消息只被其中一个消费者收到不会重复。这是我们选队列而不是主题的关键语义。3.4 主题模式的持久订阅到底怎么玩队列讲完讲主题。主题和队列的最大区别前面已经说了一份消息会发给所有订阅者。但这里有个经典痛点如果消费者用的是普通订阅也就是非持久订阅那么它离线期间的消息就收不到。很多初学者在这里翻车以为消费者启动就能收到所有消息。要让主题消费者在离线期间也不丢消息要用 JMS 的持久订阅。写法上需要做两件事先给 Connection 设置一个全局唯一的 clientId然后调用session.createDurableSubscriber(topic, 订阅名)而不是createConsumer(topic)。Connection connection factory.createConnection(); connection.setClientID(sub-001); connection.start(); Session session connection.createSession(false, Session.AUTO_ACKNOWLEDGE); Topic topic session.createTopic(topic.demo); // 第二个参数是持久订阅名称必须设置 MessageConsumer consumer session.createDurableSubscriber(topic, durable-sub-001);先启动这个消费者跑一次让 broker 记住这个持久订阅然后停掉它。接下来发几条消息再重新启动消费者会发现离线期间的消息被补回来了。持久订阅有两点要注意。一是 clientId 在同一个 broker 里必须全局唯一如果两个连接抢同一个 clientId后一个连上来会被直接拒绝。二是持久订阅需要先注册一次broker 才知道要为你保留消息如果你从头到尾只用createConsumer永远不会有补发效果。提示clientId 在同一 broker 中必须唯一否则后建立的连接会被拒绝。持久订阅名也不要随意变否则会生成多个订阅副本。这个特性在实际项目中很有用但也意味着 broker 要为每个持久订阅保存消息直到消费者确认。如果订阅者长期不上线消息容易堆积部署时要想好清理策略。控制台 Topics 页面能看到每个订阅者的状态是排查这类问题的入口。4. 可靠性与实战排错这些坑我基本都踩过4.1 消息确认机制和事务会话消息确认是 JMS 学习里最绕、也最不能跳过的部分。Session 创建时的第二个参数就是确认模式。AUTO_ACKNOWLEDGE是最简单的自动确认消费者拿到消息就算确认完成broker 可以把消息从队列里删掉或者不再投递。适合大多数对丢失不敏感的业务。CLIENT_ACKNOWLEDGE则把确认动作交给业务代码你需要主动调用message.acknowledge()。这样做的好处是一条消息只有当业务逻辑真正处理成功了才算消费完成如果中途出错、没确认就断开连接broker 会把这条消息重新投递。要注意的是CLIENT_ACKNOWLEDGE 模式下调用 acknowledge 是会话级别的会把这个 Session 里已经接收但还没确认的消息一次性确认掉不是只确认那一条。这点我最初理解错了以为是每条消息单独确认后来看文档仔细琢磨才明白。注意CLIENT_ACKNOWLEDGE 模式下acknowledge() 确认的是这个 Session 里当前累积的所有未确认消息而不是只确认你手里那一条。理解这一点才能避免“我以为确认了其实把不该确认的都确认了”的误操作。还有一个DUPS_OK_ACKNOWLEDGE允许消息重复投递性能好但要求消费者能容忍重复数据。实际场景中我宁可选择显式确认也不愿意在业务里去费劲做幂等除非性能压力特别大。如果开了事务会话做法又不一样。createSession(true, Session.AUTO_ACKNOWLEDGE)之后发送消息后必须调用session.commit()才真正入队消费完后也要commit()表示这批消息被确认。中途想反悔就rollback()。事务会话把多条消息的发送和确认包在一个原子操作里适合需要保证一组操作一致性的业务。4.2 消息持久化与丢失风险默认情况下ActiveMQ 会把消息持久化到磁盘broker 重启后消息不丢。默认的持久化方案叫KahaDB存储目录在data/kahadb下面。你往队列里发了几条持久化消息重启服务后仍然能看到它们靠的就是 KahaDB。如果不想要这样的可靠性可以在生产者发送时使用DeliveryMode.NON_PERSISTENT消息只放在内存里性能更好但 broker 一重启就没了。适合日志、实时统计这类丢了可以接受的场景。代码里就是producer.send(message, DeliveryMode.NON_PERSISTENT, 4, 0L)这样的重载第二个参数指定模式第三个是优先级第四个是过期时间。生产环境也可以在activemq.xml的persistenceAdapter配置里换成 JDBC 持久化把消息存到数据库。这通常是为了跟已有业务数据共用一套运维体系但写入吞吐会下降。具体怎么选要结合消息量和数据库的依赖情况权衡。学习阶段先知道有这几种方案就行。需要说明的是持久化本身解决的是“broker 重启后消息还在”的问题它不能解决“消费者处理逻辑出错导致的消息处理失败”。后者要靠确认机制和重投递来解决。4.3 JMS 学习中的高频问题速查学习过程中我列过一张排查表基本都是踩过或帮朋友排查过的问题直接上表现象常见原因排查与解决连接被拒绝broker 没启动、连错端口、防火墙拦截看 61616 端口是否 LISTEN确认连接地址是 tcp:// 协议消费者收不到消息忘了 connection.start()、队列里没有消息、消费类型选错先看控制台队列是否有积压再检查代码有没有 startTopic 消息丢失非持久订阅且消费者离线或发布在前订阅在后用持久订阅先注册订阅名再发消息订阅者收到重复数据多个消费者都订阅了 Topic确认业务到底该用 Queue 还是 Topic控制台登录不了默认账号密码被改过或 user.properties 配置问题检查 conf 目录账号配置服务端日志也有提示消息处理失败反复重试确认模式下没正确 acknowledge检查代码是否主动确认事务里是否 commit依赖冲突导致 ClassNotFoundactivemq-all 与项目里某些库版本冲突查看依赖树换精简依赖比如 activemq-client消息大量堆积消费速度跟不上生产速度看消费者数量、是否使用异步监听、是否有消费端挂掉这张表解决了绝大多数问题。还有一个容易被忽略的细节队列里消费不掉的消息会被 ActiveMQ 移到一个叫ActiveMQ.DLQ的死信队列控制台 Queues 页面能看到。看到 DLQ 里有数据不用慌那其实是排查线索比如消息格式问题或者重试次数超限。4.4 给同样在学 JMS 的人几点建议按照我这轮学习下来的顺序给几个实操性强的建议。第一先从队列跑通再碰主题。队列模式简单直接能最快把七个核心对象串起来。主题上的持久订阅、订阅名、clientId 都是绕弯的知识放到第二阶段去啃。第二先同步接收再异步监听。同步receive带超时时间的写法很容易复现和理解异步监听虽然代码简洁但回调线程、异常处理这些细节对新手不友好。第三验证问题别只靠猜多用控制台。ActiveMQ 控制台能看到队列积压、消费者在线状态、消息数量很多问题一眼就能定位。第四学习 JMS 不只是在学 ActiveMQ。JMS 这套模型是消息中间件思想的标准抽象你弄懂了生产者、消费者、队列、订阅、确认、持久化这些概念后面接触其他消息服务会快很多。我最大的感受就是以前看消息中间件的文档总感觉隔了一层现在再看它们的概念第一反应是“这其实就是 JMS 模型下的一种改动”。最后学任何中间件都要动手跑一遍。纸上谈兵永远分不清“consumer 没 start 收不到消息”和“broker 挂了连接失败”到底有什么区别。把这套环境跑通后面很多问题自己就有感觉了。