从我自己的经历说起吧。在接手一个内部系统重构时需要把耗时较长的报表生成任务从请求线程里摘出去后端不用一直傻等前端也不用一直转圈。当时组里讨论了好几个方案最后敲定用 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 Classic5.x是传统版本功能完整成熟稳定我项目里用的就是这套。ActiveMQ Artemis2.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 中转”立刻有了直观认知。如果启动报端口占用通常是因为 61616OpenWire 端口或 8161Web 控制台被占用修改 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 usernameapp-user passwordapp-pass groupsusers/ /users /simpleAuthenticationPlugin /plugins这些配置不复杂但每一条都可能直接影响线上稳定性。改完配置文件都需要重启 Broker 才生效测试环境改完务必验证一遍。4. 动手实战Producer 和 Consumer 的关键代码与运行细节本地环境准备好了就可以开始写代码了。我先用 Maven 建一个普通 Java 工程引入 ActiveMQ 客户端依赖。版本号要和 Broker 版本匹配我用的版本组合是 ActiveMQ 5.18.x 对应客户端依赖 5.18.x不要混用大版本否则可能出现协议兼容问题。dependency groupIdorg.apache.activemq/groupId artifactIdactivemq-client/artifactId version5.18.2/version /dependency4.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机制。它要求消费者必须设置一个持久化客户端 IDBroker 会为它保留离线期间的消息等它重新上线后再推送。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 协议让客户端能在连接断开后自动重连到 BrokerString brokerURL failover:(tcp://localhost:61616)?initialReconnectDelay1000maxReconnectAttempts20; ActiveMQConnectionFactory factory new ActiveMQConnectionFactory(brokerURL);failover 协议是 ActiveMQ 留给生产环境的“保命”配置它能在底层连接断开时按照参数配置重连。配合异步监听模式重连后消费者会自动恢复订阅业务代码几乎无感。6.3 手动确认模式下忘记 acknowledge 导致消息永远收不到新消息用 CLIENT_ACKNOWLEDGE 模式时如果消费者拉取消息后代码抛异常没有调用 acknowledgeBroker 就不会移除消息。但这里有个隐蔽问题如果你每次连的是同一条消息一直拉取失败那后续新消息就进不来因为前面的消息堵住了队列头。换一批新消息之前要确保每一条都成功确认或者用 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 挂了数据不丢”。把这些机制放到真实业务里去理解代码怎么写都顺手。