后端干久了谁还没被几把异步入门折腾过。A系统要调B系统B响应慢A的线程就挂在那里等流量一冲上来数据库连接池先被打满紧接着整条链路一起雪崩。不少团队这时候的第一反应是“中间塞一个消息队列”。Java生态里聊消息队列最绕不开的组合就是JMS规范和它的老牌实现ActiveMQ。这篇标题叫“JMS, ActiveMQ 学习一则搜”说白了就是一份学习过程中的速查手册把JMS标准里的核心概念、ActiveMQ的常用玩法、还有我在生产环境里踩过的坑整理成边学边查的经验笔记。如果你是刚接触消息中间件的Java开发者是项目里已经用了ActiveMQ但还没系统理过JMS API的工程师或者想快速温习消息机制的面试党这篇应该帮得上忙。1. JMS是什么先分清规范、实现与边界1.1 JMS规范到底解决了什么问题在JMS出现之前Java程序想收发消息基本被某一种消息中间件的私有客户端API绑死。你写了A厂商的代码将来想换B厂商所有调用细节几乎都要重写更麻烦的是连接管理、会话、消息确认这些机制各家都不一样团队里每个人各学各的经验完全没法沉淀。JDBC之所以能一统数据库访问核心思路就是暴露统一标准接口、把实现交给驱动的思想。JMS走的是同一条路它把Connection、Session、MessageProducer、MessageConsumer这些概念抽象成Java标准API并定义了它们的行为底层由ActiveMQ这样的消息中间件去实现。所以业务代码里依赖的是javax.jms包里的接口真正干活的是Broker端这个边界必须分清不然看到代码里混着ActiveMQ的扩展API很容易把“JMS标准”和“中间件实现”搅成一锅粥。1.2 ActiveMQ在生态里的位置与适用场景ActiveMQ是JMS规范一个非常经典的开源实现。它功能够全面持久化、事务、XA、多协议接入、管理控制台全都有文档也攒了足够多年头Java社区接受度很高很多中小型项目的消息选型都落在它身上。一句到位的话JMS是标准ActiveMQ是这套标准下的产物。代码里看到的Queue、MessageConsumer是JMS接口而ActiveMQConnectionFactory、failover协议这些则是ActiveMQ的实现与扩展。实际业务里ActiveMQ最常解决三类问题系统解耦、异步提速、削峰填谷。举个例子下单后要把订单信息同步给库存、积分、通知三个系统如果同步调用串行等待的时间很长用一个队列发出去各系统自己到队列里取消息互不阻塞这就是解耦与异步。秒杀场景下瞬时流量很大订单入口先把请求写进队列库存服务按自己能力慢慢扣减这是削峰。把这些使用边界想清楚学习的时候就不容易一头扎进API细节里出不来。2. 核心模型拆解连接、会话、目的地不能含糊2.1 从ConnectionFactory到Session的完整链路JMS的编程模型仔细看就是一条固定链路创建ConnectionFactory、创建Connection、创建Session、拿到Destination、再创建Producer或者Consumer最后发送或接收消息。很多新手第一个疑惑是为什么要这么绕不能直接new一个Queue然后发消息吗答案是不能因为JMS要让连接、会话、生产者消费者全部走标准化的生命周期管理。ConnectionFactory负责创建到Broker的物理连接内部其实就是在维护TCP连接和线程池Connection代表一个连接Session代表在这个连接上进行消息收发的工作单元Destination就是消息要去的地方队列也好、主题也好。一个容易忽略的点是Connection需要主动调用start()后消费者才会开始接收消息。生产者不start也能发送因为消息发送走的是底层连接通道但消费者端必须等连接进入启动状态消息才会投递给你。我见过不少新手只盯着日志半天收不到消息最后发现就是忘了调connection.start()。下面是一段最简生产者的代码省略了异常处理ActiveMQConnectionFactory factory new ActiveMQConnectionFactory(tcp://127.0.0.1:61616); Connection connection factory.createConnection(); connection.start(); Session session connection.createSession(false, Session.AUTO_ACKNOWLEDGE); Queue queue session.createQueue(ORDER_QUEUE); MessageProducer producer session.createProducer(queue); TextMessage msg session.createTextMessage(hello jms); producer.send(msg);Connection属于重量级对象整个应用复用一份就好频繁创建关闭会平白增加大量握手开销Session相对轻量可以在同一个Connection下创建多个按业务模块划分使用。2.2 Queue与Topic选错模型会出大问题JMS里目的地有两种Queue和Topic语义完全不同选错会在业务上出大问题。Queue是点对点模型。一条消息只会被一个消费者消费。消息发送到Queue后broker保存它直到某个消费者取走并确认处理完成。哪怕同时挂了10个消费者一条消息也只会交给其中一个处理。这种模型天然适合任务分发一批订单、一批短信大家分散处理处理完了消息也就没了。Topic是发布订阅模型。一条消息会被所有订阅了该主题的消费者都收到。适合广播通知系统公告、状态变更事件想让所有关心的人都知道就用Topic。这里有个大坑如果消息发布时某个Topic订阅者不在线非持久订阅者会直接错过这条消息不会等它回来补偿。用表格列一下关键区别对比项QueueTopic消息被谁消费一个消费者所有订阅者各收一份消费后消息确认后移除订阅者各自偏移离线订阅者上线后可以继续消费非持久订阅会丢失典型场景任务队列、工作分配事件广播、通知我的经验是任务型处理优先考虑Queue事件型、需要多方感知的才用Topic。不少团队上来闭眼用Queue等做通知广播时每条消息只有一个人收到才发现模型选错了。2.3 消息组成与Body类型的取舍一条JMS消息由三部分组成消息头、属性、消息体。消息头里是一些标准化字段比如JMSDestination表示发往哪里、JMSMessageID是消息唯一编号、JMSDeliveryMode区分持久或非持久、JMSExpiration可以设置过期时间、JMSPriority是优先级。这些字段由JMS规范统一命名使用中间件时不建议随便改动尤其是JMSExpiration和JMSDeliveryMode直接关系消息是否会被丢弃、是否会被持久化。属性部分可以用来存放自定义的“路由标签”例如订单来源、业务类型消费者可以根据属性做筛选。但这个筛选是broker端按消息选择器执行的语法有点像SQL里的where条件用得多会消耗broker性能尽量只放真正需要检索的字段。消息体有几种类型使用频率差异很大。TextMessage最常用就是装一段字符串MapMessage可以放一组键值对BytesMessage适合二进制数据ObjectMessage包装一个Java对象用起来方便但序列化版本兼容性和反序列化安全都需要注意跨系统传输时尤其要小心我一般更推荐直接序列化成JSON后放到TextMessage里。另外超出几MB的大消息不要塞消息中间件对broker内存和存储压力都很大真遇到大对象先把内容放到文件或者对象存储消息里只带一个访问标识。2.4 消息确认机制四种ACK模式的坑消息确认ACK决定一条消息何时算“处理成功”。JMS里有4种模式选错直接影响消息会不会丢。AUTO_ACKNOWLEDGE是最常用也最省心的模式消费者收到消息在消息处理回调返回前底层自动向broker确认。但这里藏着一个经典坑如果你的业务逻辑在接收消息之外又异步去做比如拿到消息后丢到一个线程池里处理AUTO_ACKNOWLEDGE下消息会被提前确认线程池里如果处理失败这条消息就丢了。消息丢失不是broker吞掉的是你确认得太早。CLIENT_ACKNOWLEDGE把确认权交给你消费者收到后必须手动调用message.acknowledge()broker才会认为处理成功。适合业务逻辑复杂、必须等整个流程跑完再确认的场景。要注意acknowledge()确认的是这条session里当前以及之前收到的所有消息不是只确认你手里这一条这个语义别记错了。DUPS_OK_ACKNOWLEDGE是宽松模式允许重复确认吞吐高但可能重复投递。SESSION_TRANSACTED则是配合事务会话使用由session.commit()统一确认一批消息。四种模式里我实际用到最多的是AUTO和CLIENT。追求简单可以AUTO但前提是消费者拿到消息后必须在回调里同步处理完涉及写库、调外部系统这类必须保证不丢的场景老老实实用CLIENT_ACKNOWLEDGE。3. 实操部分搭一个能跑的ActiveMQ并收发消息3.1 下载安装与Broker启动这一节讲实际操作。从官网下载ActiveMQ的二进制包选稳定版本不要追新解压到本地目录。在bin目录下Unix环境执行./activemq startWindows环境执行activemq.bat startBroker就起来了。默认会监听61616端口用于消息协议8161端口是Web管理控制台。启动后几个确认手段查看日志文件确认无异常浏览器打开8161端口能看到管理控制台用netstat确认61616端口处于监听状态。控制台里能看到Queue、Topic的实时消息数、消费者数量还能发测试消息开发调试阶段非常顺手。有几个配置上的默认坑。一是默认的用户密码是固定的部署到任何非本机环境前都要改掉尤其管理控制台二是JVM内存参数默认偏小生产环境要单独调大堆内存可以在启动脚本里通过环境变量设置单位用MB还是GB要算好三是端口被占用时启动会失败先检查61616和8161是否被其他进程占用。activemq.xml是所有核心配置汇集的入口包括持久化目录、传输连接器、安全管理等后续改动基本都在这个文件里完成改动后要重启Broker才生效。3.2 生产者与消费者完整例子继续用一个能直接跑的示例讲清楚收发逻辑。生产者已经在前面给出最简版这里补上资源释放和持久化设置ActiveMQConnectionFactory factory new ActiveMQConnectionFactory(tcp://127.0.0.1:61616); try (Connection connection factory.createConnection()) { Session session connection.createSession(false, Session.AUTO_ACKNOWLEDGE); Queue queue session.createQueue(ORDER_QUEUE); MessageProducer producer session.createProducer(queue); producer.setDeliveryMode(DeliveryMode.PERSISTENT); for (int i 0; i 100; i) { producer.send(session.createTextMessage(order- i)); } }这里用了try-with-resources连接关闭时session和producer会一并释放比手写close可靠。消费者分同步和异步两种写法。同步写法用一个循环调用consumer.receive()会阻塞等待消息适合简单单线程测试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); while (true) { TextMessage message (TextMessage) consumer.receive(3000); if (message null) { continue; } System.out.println(receive: message.getText()); }异步写法是注册一个MessageListener回调broker把消息推给客户端线程时自动触发。实际项目里异步监听更主流消费者可以一直待命consumer.setMessageListener(message - { try { TextMessage tm (TextMessage) message; System.out.println(async receive: tm.getText()); } catch (JMSException e) { e.printStackTrace(); } });注意监听器方法里异常只能捕获不能“吞掉”后当无事发生如果业务处理失败要考虑结合CLIENT_ACKNOWLEDGE重新确认或者让异常上抛触发broker重投策略。3.3 事务Session与批量发送createSession的第一个参数代表是否开启事务。如果传true第二个参数会被忽略所有send和receive操作都进入一个事务单元直到调用session.commit()才会真正生效调用rollback()则撤销当前批次。事务会话的最大价值有两个批量发送性能和原子性。批量发送1000条消息如果每条send后面都单独commit性能很差改成收集一批后统一commit网络往返大幅减少。消费端的批量确认同理一次消费几百条后commit一次比每条确认高效得多。原子性方面事务内发送多条消息要么全部成功要么全部失败适合多消息必须一起到达的场景。但这里有个容易翻车的点事务会话中消息发出后在还没commit时本会话内receive是可以看到自己刚发的那条消息的但其他消费者不会。很多人在事务里发消息后马上起另一个消费者去验证结果死活收不到就是因为忘了commit。另外消费端如果一直不commitbroker会认为消息还在处理中不会分给别的消费者这既是保护也是负担事务不能长时间挂着不提交。3.4 持久化与基本配置默认的持久化存储是kahaDB数据目录在data/kahadb下。所谓持久化消息是指发送时设置了DeliveryMode.PERSISTENT的消息broker会先写入磁盘存储再返回发送成功Broker重启后这些消息不会丢。非持久化消息只存在于内存重启即消失。持久化的意义要看业务。订单通知、交易事件这类绝不能丢的用PERSISTENT日志、瞬时指标这类允许丢失的用NON_PERSISTENT能换来更高吞吐。持久化机制本身也有开销不需要时不要盲目全量持久化。activemq.xml里可以调整持久化适配器、内存限制、连接器地址。我实际碰到的教训是不要为了省事把数据目录放到临时目录下一旦系统清理重启所有持久化消息都蒸发了。数据目录最好单独挂盘容量规划按消息一天的量乘上保留天数粗算再留50%的余量。4. 生产环境排错记录消息丢失、堆积与断连4.1 消息丢了先查ACK再查持久化线上消息丢失我排查时习惯按这个顺序走。第一先确认消息是不是持久化消息如果是NON_PERSISTENT且Broker重启过丢失是预期行为应用自己要能接受这种语义。第二看消费者的确认模式AUTO_ACKNOWLEDGE加上异步线程池处理是最常见的“丢消息”元凶修复办法是改成CLIENT_ACKNOWLEDGE并在处理成功后确认。第三看是否有过期时间JMSExpiration被设置且消费者消费不及时消息会过期被清理。第四看异常处理监听器里捕获异常后什么都不做broker会默认确认这也相当于丢。这四条检查完九成的“消息丢了”问题都能定位。注意不要一上来就怀疑Broker丢消息KahaDB的持久化机制本身是很可靠的绝大多数情况是使用端的确认语义出了问题。4.2 消息堆积预取、消费者数量与死信队列消息堆积的表象是Queue里pending消息数持续增长、消费延迟变大但根源要仔细分辨。第一种是生产速度确实太快下游处理不过来这种是容量问题优先加消费者实例增加消费并发。JMS里一个Queue可以挂多个消费者并发拉取单机可以加线程多机可以加节点。第二种是某个消费者卡住了但它还在线导致消息被它预取走。默认prefetchSize是1000消费者会一次性把上千条消息拉到本地缓冲如果它处理很慢其他消费者明明闲着也拿不到这部分消息。解决思路是调小prefetch比如改成100、10让消息更均匀地分给多个消费者。这个参数在ActiveMQ连接URL里设置tcp://127.0.0.1:61616?jms.prefetchPolicy.queuePrefetch100。还有一种堆积表现在ActiveMQ.DLQ死信队列。消息反复消费失败后会被移入死信队列默认的死信队列名是ActiveMQ.DLQ。如果业务里看到这个队列有数据说明有消息一直处理不成功要查消费端逻辑和依赖的服务而不是把死信消息重新塞回原队列重试那只会反复失败反复进死信。我遇到过因为反序列化类缺失导致大批消息进死信的情况排查方法就是打开死信队列看消息内容一眼就能看到堆栈。4.3 断线重连与failover配置生产环境里网络不可能永远稳定。客户端连着Broker一条TCP断掉如果客户端实现没配重连整个消费线程就静默挂死消息越积越多也没人处理。ActiveMQ客户端支持failover协议做断线重连。连接地址写成failover:(tcp://host1:61616,tcp://host2:61616)?randomizefalse这样Broker单点挂了客户端会自动切换到另一个节点。randomizefalse是让客户端按顺序连接随机模式在多节点下会分散负载各有适用场景。failover还能配合超时参数failover:(tcp://host1:61616)?timeout3000超时后重连失败会抛异常避免无限等待。客户端进程里还可以设置心跳参数让客户端定时确认Broker在线发现失联后更快触发重连。排查断线问题时我常先看Broker端日志里的连接关闭记录再对比客户端日志判断是Broker主动断还是网络层断。有一点要注意failover不是万能的。如果Broker在消费过程中宕机消息可能被重复投递客户端要做好幂等如果消息已提交但还没ack重启后会被重新投递这也是重复消费的来源之一。4.4 重复消费要做好最坏打算消息中间件普遍是at-least-once语义也就是一条消息可能被投递不止一次。你在网上能看到各种“刚好一次”的讨论但在ActiveMQ这类经典JMS实现里靠客户端配置实现严格不重复很难最稳妥的办法是消费端自己保证幂等。我的习惯是凡是写库、扣库存、发短信这些操作业务上要有唯一键约束。消费时先以消息里带的业务唯一ID去查一下或者插入一条去重记录重复消息直接忽略。比如用数据库的唯一索引、Redis的SETNX都能低成本挡住重复消费。不要把希望寄托在“应该不会重复”生产环境里复议、人工干预、重放消息都是常态没有幂等意味着每次重放都可能造成资损。5. 选型思考与影响范围ActiveMQ什么时候该上5.1 和几类主流中间件的取舍ActiveMQ不是唯一选择但它适合的范围很明确。关注点ActiveMQ流式处理产品如KafkaRabbitMQ典型定位企业级消息服务、事务海量日志、流计算灵活路由、生态协议多语言生态Java体系体验最好多语言偏大数据多语言路由能力强消息模型标准Queue/Topic分区消费组ExchangeQueue运维成本适中集群规模大时偏高适中事务/可靠性支持到位较弱侧重吞吐一般如果团队是Java技术栈、需求集中在业务系统解耦、异步通知、跨系统事件ActiveMQ学习曲线平缓JMS标准写出来的代码还有迁移空间是稳的选择。如果是每秒百万级日志摄取、数据管道式处理ActiveMQ的吞吐不是它的强项流式处理平台更合适。如果业务里需要特别复杂的路由规则RabbitMQ的Exchange模型会更灵活。选型别只看性能先看团队能不能维护、业务模型合不合适。我在项目里见过为了“高端”强行引一套复杂组件结果半年没人会配置最后反而回退到ActiveMQ的情况。5.2 一个典型异步链路的参数估算用一个场景估算方便知道ActiveMQ的容量大致怎么算。假设订单系统下单后要异步通知短信服务、积分服务、库存服务峰值每秒下单500笔每笔产生一条JSON消息消息体平均2KB。目标是把消费延迟控制在5秒以内。第一步算每秒消息量500条每条2KB每秒1MB的数据量带宽和内存都不是瓶颈。第二步看消费能力假设单线程消费者处理一条消息平均20ms一个消费者每秒只能处理50条要撑住500条/秒的峰值至少需要10个消费者线程考虑实际CPU抢占、GC停顿按照0.7的利用率系数建议配14到15个消费者线程。第三步算队列容量5秒延迟对应队列里积压2500条消息每条约2KB总共5MB任何机器都扛得住。这个估算说明ActiveMQ处理这种量级绰绰有余瓶颈通常在消费端业务逻辑不在Broker端。这类估算到这里就够了不必做得很精细。真到瓶颈时再去调prefetch、消费者并发、批量确认方向清楚知道下一步动哪里。6. 学习路线与压箱底建议6.1 值得照做的学习顺序我见过不少人在ActiveMQ上学习路线很拧巴先啃activemq.xml各种高级配置用一堆特有API结果JMS最基础的模型反而没吃透。正确的顺序应该反过来。第一步是彻底吃透JMS核心模型ConnectionFactory、Connection、Session、Destination、Producer、Consumer这六个角色的关系和生命周期最好能不看文档画出调用链这比记住任何配置都值钱。第二步是用最简代码跑通Queue和Topic两种模型的收发把ACK、持久化、事务每一个点都实践一遍体会每种模式对消息生命周期的影响。第三步才是看ActiveMQ的特性failover、虚拟目的地、消息选择器、监控接口这些是使用中的加分项。第四步建议看Broker配置和数据目录结构理解持久化、内存、日志、管控台到这一步你对ActiveMQ就有比较完整的认知了。反过来学一上来陷入细节很容易三个月只记住一堆参数遇到问题还是不会排查。6.2 几个容易忽略的注意事项最后把我压箱底的经验一次写完。Session不是线程安全的。同一个Session里的生产者、消费者、监听器不要在多个线程同时用多线程需要各自创建Session但Connection可以共享。很多人写并发消费时共享了一个Session结果时不时报错就是这个原因。MessageProducer和MessageConsumer都可以复用不用每条消息都新建但连接和会话的关闭时序要小心。try-with-resources里先关生产者、消费者最后关Connection顺序反了会导致资源释放不彻底。管理控制台里看到的消息数量不等于实际即时状态。控制台有刷新延迟只适合人工观察别拿它做监控告警数据源监控要用JMX或统计接口。ObjectMessage能不用就不用。跨语言、跨版本、外部客户端的场景下序列化问题能缠死你统一转JSON字符串是更省心的路子。broker嵌入模式。做单元测试时可以在测试里启动一个嵌入式Broker代码里创建BrokerService并start不需要外部进程测试稳定性和速度都很理想这也是我调试时最顺手的一招。最后再分享一个细节我每次交付ActiveMQ相关项目都会留下一份速查笔记把连接URL、关键配置、排错路径、幂等策略写清楚。因为ActiveMQ本身不难难的是半年后再让人接手时所有坑都重新踩一遍。“学习一则搜”这个标题其实就是想说与其把经验留在脑子里不如整理成一份随查随用的手册下次遇到问题翻自己的笔记比重新翻文档快得多。 
长河君 · 2026/10/20 6:55:33 · 阅读 1.2k 深入浅出JMS与ActiveMQ：消息中间件核心概念与生产环境实践
后端干久了谁还没被几把异步入门折腾过。A系统要调B系统B响应慢A的线程就挂在那里等流量一冲上来数据库连接池先被打满紧接着整条链路一起雪崩。不少团队这时候的第一反应是“中间塞一个消息队列”。Java生态里聊消息队列最绕不开的组合就是JMS规范和它的老牌实现ActiveMQ。这篇标题叫“JMS, ActiveMQ 学习一则搜”说白了就是一份学习过程中的速查手册把JMS标准里的核心概念、ActiveMQ的常用玩法、还有我在生产环境里踩过的坑整理成边学边查的经验笔记。如果你是刚接触消息中间件的Java开发者是项目里已经用了ActiveMQ但还没系统理过JMS API的工程师或者想快速温习消息机制的面试党这篇应该帮得上忙。1. JMS是什么先分清规范、实现与边界1.1 JMS规范到底解决了什么问题在JMS出现之前Java程序想收发消息基本被某一种消息中间件的私有客户端API绑死。你写了A厂商的代码将来想换B厂商所有调用细节几乎都要重写更麻烦的是连接管理、会话、消息确认这些机制各家都不一样团队里每个人各学各的经验完全没法沉淀。JDBC之所以能一统数据库访问核心思路就是暴露统一标准接口、把实现交给驱动的思想。JMS走的是同一条路它把Connection、Session、MessageProducer、MessageConsumer这些概念抽象成Java标准API并定义了它们的行为底层由ActiveMQ这样的消息中间件去实现。所以业务代码里依赖的是javax.jms包里的接口真正干活的是Broker端这个边界必须分清不然看到代码里混着ActiveMQ的扩展API很容易把“JMS标准”和“中间件实现”搅成一锅粥。1.2 ActiveMQ在生态里的位置与适用场景ActiveMQ是JMS规范一个非常经典的开源实现。它功能够全面持久化、事务、XA、多协议接入、管理控制台全都有文档也攒了足够多年头Java社区接受度很高很多中小型项目的消息选型都落在它身上。一句到位的话JMS是标准ActiveMQ是这套标准下的产物。代码里看到的Queue、MessageConsumer是JMS接口而ActiveMQConnectionFactory、failover协议这些则是ActiveMQ的实现与扩展。实际业务里ActiveMQ最常解决三类问题系统解耦、异步提速、削峰填谷。举个例子下单后要把订单信息同步给库存、积分、通知三个系统如果同步调用串行等待的时间很长用一个队列发出去各系统自己到队列里取消息互不阻塞这就是解耦与异步。秒杀场景下瞬时流量很大订单入口先把请求写进队列库存服务按自己能力慢慢扣减这是削峰。把这些使用边界想清楚学习的时候就不容易一头扎进API细节里出不来。2. 核心模型拆解连接、会话、目的地不能含糊2.1 从ConnectionFactory到Session的完整链路JMS的编程模型仔细看就是一条固定链路创建ConnectionFactory、创建Connection、创建Session、拿到Destination、再创建Producer或者Consumer最后发送或接收消息。很多新手第一个疑惑是为什么要这么绕不能直接new一个Queue然后发消息吗答案是不能因为JMS要让连接、会话、生产者消费者全部走标准化的生命周期管理。ConnectionFactory负责创建到Broker的物理连接内部其实就是在维护TCP连接和线程池Connection代表一个连接Session代表在这个连接上进行消息收发的工作单元Destination就是消息要去的地方队列也好、主题也好。一个容易忽略的点是Connection需要主动调用start()后消费者才会开始接收消息。生产者不start也能发送因为消息发送走的是底层连接通道但消费者端必须等连接进入启动状态消息才会投递给你。我见过不少新手只盯着日志半天收不到消息最后发现就是忘了调connection.start()。下面是一段最简生产者的代码省略了异常处理ActiveMQConnectionFactory factory new ActiveMQConnectionFactory(tcp://127.0.0.1:61616); Connection connection factory.createConnection(); connection.start(); Session session connection.createSession(false, Session.AUTO_ACKNOWLEDGE); Queue queue session.createQueue(ORDER_QUEUE); MessageProducer producer session.createProducer(queue); TextMessage msg session.createTextMessage(hello jms); producer.send(msg);Connection属于重量级对象整个应用复用一份就好频繁创建关闭会平白增加大量握手开销Session相对轻量可以在同一个Connection下创建多个按业务模块划分使用。2.2 Queue与Topic选错模型会出大问题JMS里目的地有两种Queue和Topic语义完全不同选错会在业务上出大问题。Queue是点对点模型。一条消息只会被一个消费者消费。消息发送到Queue后broker保存它直到某个消费者取走并确认处理完成。哪怕同时挂了10个消费者一条消息也只会交给其中一个处理。这种模型天然适合任务分发一批订单、一批短信大家分散处理处理完了消息也就没了。Topic是发布订阅模型。一条消息会被所有订阅了该主题的消费者都收到。适合广播通知系统公告、状态变更事件想让所有关心的人都知道就用Topic。这里有个大坑如果消息发布时某个Topic订阅者不在线非持久订阅者会直接错过这条消息不会等它回来补偿。用表格列一下关键区别对比项QueueTopic消息被谁消费一个消费者所有订阅者各收一份消费后消息确认后移除订阅者各自偏移离线订阅者上线后可以继续消费非持久订阅会丢失典型场景任务队列、工作分配事件广播、通知我的经验是任务型处理优先考虑Queue事件型、需要多方感知的才用Topic。不少团队上来闭眼用Queue等做通知广播时每条消息只有一个人收到才发现模型选错了。2.3 消息组成与Body类型的取舍一条JMS消息由三部分组成消息头、属性、消息体。消息头里是一些标准化字段比如JMSDestination表示发往哪里、JMSMessageID是消息唯一编号、JMSDeliveryMode区分持久或非持久、JMSExpiration可以设置过期时间、JMSPriority是优先级。这些字段由JMS规范统一命名使用中间件时不建议随便改动尤其是JMSExpiration和JMSDeliveryMode直接关系消息是否会被丢弃、是否会被持久化。属性部分可以用来存放自定义的“路由标签”例如订单来源、业务类型消费者可以根据属性做筛选。但这个筛选是broker端按消息选择器执行的语法有点像SQL里的where条件用得多会消耗broker性能尽量只放真正需要检索的字段。消息体有几种类型使用频率差异很大。TextMessage最常用就是装一段字符串MapMessage可以放一组键值对BytesMessage适合二进制数据ObjectMessage包装一个Java对象用起来方便但序列化版本兼容性和反序列化安全都需要注意跨系统传输时尤其要小心我一般更推荐直接序列化成JSON后放到TextMessage里。另外超出几MB的大消息不要塞消息中间件对broker内存和存储压力都很大真遇到大对象先把内容放到文件或者对象存储消息里只带一个访问标识。2.4 消息确认机制四种ACK模式的坑消息确认ACK决定一条消息何时算“处理成功”。JMS里有4种模式选错直接影响消息会不会丢。AUTO_ACKNOWLEDGE是最常用也最省心的模式消费者收到消息在消息处理回调返回前底层自动向broker确认。但这里藏着一个经典坑如果你的业务逻辑在接收消息之外又异步去做比如拿到消息后丢到一个线程池里处理AUTO_ACKNOWLEDGE下消息会被提前确认线程池里如果处理失败这条消息就丢了。消息丢失不是broker吞掉的是你确认得太早。CLIENT_ACKNOWLEDGE把确认权交给你消费者收到后必须手动调用message.acknowledge()broker才会认为处理成功。适合业务逻辑复杂、必须等整个流程跑完再确认的场景。要注意acknowledge()确认的是这条session里当前以及之前收到的所有消息不是只确认你手里这一条这个语义别记错了。DUPS_OK_ACKNOWLEDGE是宽松模式允许重复确认吞吐高但可能重复投递。SESSION_TRANSACTED则是配合事务会话使用由session.commit()统一确认一批消息。四种模式里我实际用到最多的是AUTO和CLIENT。追求简单可以AUTO但前提是消费者拿到消息后必须在回调里同步处理完涉及写库、调外部系统这类必须保证不丢的场景老老实实用CLIENT_ACKNOWLEDGE。3. 实操部分搭一个能跑的ActiveMQ并收发消息3.1 下载安装与Broker启动这一节讲实际操作。从官网下载ActiveMQ的二进制包选稳定版本不要追新解压到本地目录。在bin目录下Unix环境执行./activemq startWindows环境执行activemq.bat startBroker就起来了。默认会监听61616端口用于消息协议8161端口是Web管理控制台。启动后几个确认手段查看日志文件确认无异常浏览器打开8161端口能看到管理控制台用netstat确认61616端口处于监听状态。控制台里能看到Queue、Topic的实时消息数、消费者数量还能发测试消息开发调试阶段非常顺手。有几个配置上的默认坑。一是默认的用户密码是固定的部署到任何非本机环境前都要改掉尤其管理控制台二是JVM内存参数默认偏小生产环境要单独调大堆内存可以在启动脚本里通过环境变量设置单位用MB还是GB要算好三是端口被占用时启动会失败先检查61616和8161是否被其他进程占用。activemq.xml是所有核心配置汇集的入口包括持久化目录、传输连接器、安全管理等后续改动基本都在这个文件里完成改动后要重启Broker才生效。3.2 生产者与消费者完整例子继续用一个能直接跑的示例讲清楚收发逻辑。生产者已经在前面给出最简版这里补上资源释放和持久化设置ActiveMQConnectionFactory factory new ActiveMQConnectionFactory(tcp://127.0.0.1:61616); try (Connection connection factory.createConnection()) { Session session connection.createSession(false, Session.AUTO_ACKNOWLEDGE); Queue queue session.createQueue(ORDER_QUEUE); MessageProducer producer session.createProducer(queue); producer.setDeliveryMode(DeliveryMode.PERSISTENT); for (int i 0; i 100; i) { producer.send(session.createTextMessage(order- i)); } }这里用了try-with-resources连接关闭时session和producer会一并释放比手写close可靠。消费者分同步和异步两种写法。同步写法用一个循环调用consumer.receive()会阻塞等待消息适合简单单线程测试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); while (true) { TextMessage message (TextMessage) consumer.receive(3000); if (message null) { continue; } System.out.println(receive: message.getText()); }异步写法是注册一个MessageListener回调broker把消息推给客户端线程时自动触发。实际项目里异步监听更主流消费者可以一直待命consumer.setMessageListener(message - { try { TextMessage tm (TextMessage) message; System.out.println(async receive: tm.getText()); } catch (JMSException e) { e.printStackTrace(); } });注意监听器方法里异常只能捕获不能“吞掉”后当无事发生如果业务处理失败要考虑结合CLIENT_ACKNOWLEDGE重新确认或者让异常上抛触发broker重投策略。3.3 事务Session与批量发送createSession的第一个参数代表是否开启事务。如果传true第二个参数会被忽略所有send和receive操作都进入一个事务单元直到调用session.commit()才会真正生效调用rollback()则撤销当前批次。事务会话的最大价值有两个批量发送性能和原子性。批量发送1000条消息如果每条send后面都单独commit性能很差改成收集一批后统一commit网络往返大幅减少。消费端的批量确认同理一次消费几百条后commit一次比每条确认高效得多。原子性方面事务内发送多条消息要么全部成功要么全部失败适合多消息必须一起到达的场景。但这里有个容易翻车的点事务会话中消息发出后在还没commit时本会话内receive是可以看到自己刚发的那条消息的但其他消费者不会。很多人在事务里发消息后马上起另一个消费者去验证结果死活收不到就是因为忘了commit。另外消费端如果一直不commitbroker会认为消息还在处理中不会分给别的消费者这既是保护也是负担事务不能长时间挂着不提交。3.4 持久化与基本配置默认的持久化存储是kahaDB数据目录在data/kahadb下。所谓持久化消息是指发送时设置了DeliveryMode.PERSISTENT的消息broker会先写入磁盘存储再返回发送成功Broker重启后这些消息不会丢。非持久化消息只存在于内存重启即消失。持久化的意义要看业务。订单通知、交易事件这类绝不能丢的用PERSISTENT日志、瞬时指标这类允许丢失的用NON_PERSISTENT能换来更高吞吐。持久化机制本身也有开销不需要时不要盲目全量持久化。activemq.xml里可以调整持久化适配器、内存限制、连接器地址。我实际碰到的教训是不要为了省事把数据目录放到临时目录下一旦系统清理重启所有持久化消息都蒸发了。数据目录最好单独挂盘容量规划按消息一天的量乘上保留天数粗算再留50%的余量。4. 生产环境排错记录消息丢失、堆积与断连4.1 消息丢了先查ACK再查持久化线上消息丢失我排查时习惯按这个顺序走。第一先确认消息是不是持久化消息如果是NON_PERSISTENT且Broker重启过丢失是预期行为应用自己要能接受这种语义。第二看消费者的确认模式AUTO_ACKNOWLEDGE加上异步线程池处理是最常见的“丢消息”元凶修复办法是改成CLIENT_ACKNOWLEDGE并在处理成功后确认。第三看是否有过期时间JMSExpiration被设置且消费者消费不及时消息会过期被清理。第四看异常处理监听器里捕获异常后什么都不做broker会默认确认这也相当于丢。这四条检查完九成的“消息丢了”问题都能定位。注意不要一上来就怀疑Broker丢消息KahaDB的持久化机制本身是很可靠的绝大多数情况是使用端的确认语义出了问题。4.2 消息堆积预取、消费者数量与死信队列消息堆积的表象是Queue里pending消息数持续增长、消费延迟变大但根源要仔细分辨。第一种是生产速度确实太快下游处理不过来这种是容量问题优先加消费者实例增加消费并发。JMS里一个Queue可以挂多个消费者并发拉取单机可以加线程多机可以加节点。第二种是某个消费者卡住了但它还在线导致消息被它预取走。默认prefetchSize是1000消费者会一次性把上千条消息拉到本地缓冲如果它处理很慢其他消费者明明闲着也拿不到这部分消息。解决思路是调小prefetch比如改成100、10让消息更均匀地分给多个消费者。这个参数在ActiveMQ连接URL里设置tcp://127.0.0.1:61616?jms.prefetchPolicy.queuePrefetch100。还有一种堆积表现在ActiveMQ.DLQ死信队列。消息反复消费失败后会被移入死信队列默认的死信队列名是ActiveMQ.DLQ。如果业务里看到这个队列有数据说明有消息一直处理不成功要查消费端逻辑和依赖的服务而不是把死信消息重新塞回原队列重试那只会反复失败反复进死信。我遇到过因为反序列化类缺失导致大批消息进死信的情况排查方法就是打开死信队列看消息内容一眼就能看到堆栈。4.3 断线重连与failover配置生产环境里网络不可能永远稳定。客户端连着Broker一条TCP断掉如果客户端实现没配重连整个消费线程就静默挂死消息越积越多也没人处理。ActiveMQ客户端支持failover协议做断线重连。连接地址写成failover:(tcp://host1:61616,tcp://host2:61616)?randomizefalse这样Broker单点挂了客户端会自动切换到另一个节点。randomizefalse是让客户端按顺序连接随机模式在多节点下会分散负载各有适用场景。failover还能配合超时参数failover:(tcp://host1:61616)?timeout3000超时后重连失败会抛异常避免无限等待。客户端进程里还可以设置心跳参数让客户端定时确认Broker在线发现失联后更快触发重连。排查断线问题时我常先看Broker端日志里的连接关闭记录再对比客户端日志判断是Broker主动断还是网络层断。有一点要注意failover不是万能的。如果Broker在消费过程中宕机消息可能被重复投递客户端要做好幂等如果消息已提交但还没ack重启后会被重新投递这也是重复消费的来源之一。4.4 重复消费要做好最坏打算消息中间件普遍是at-least-once语义也就是一条消息可能被投递不止一次。你在网上能看到各种“刚好一次”的讨论但在ActiveMQ这类经典JMS实现里靠客户端配置实现严格不重复很难最稳妥的办法是消费端自己保证幂等。我的习惯是凡是写库、扣库存、发短信这些操作业务上要有唯一键约束。消费时先以消息里带的业务唯一ID去查一下或者插入一条去重记录重复消息直接忽略。比如用数据库的唯一索引、Redis的SETNX都能低成本挡住重复消费。不要把希望寄托在“应该不会重复”生产环境里复议、人工干预、重放消息都是常态没有幂等意味着每次重放都可能造成资损。5. 选型思考与影响范围ActiveMQ什么时候该上5.1 和几类主流中间件的取舍ActiveMQ不是唯一选择但它适合的范围很明确。关注点ActiveMQ流式处理产品如KafkaRabbitMQ典型定位企业级消息服务、事务海量日志、流计算灵活路由、生态协议多语言生态Java体系体验最好多语言偏大数据多语言路由能力强消息模型标准Queue/Topic分区消费组ExchangeQueue运维成本适中集群规模大时偏高适中事务/可靠性支持到位较弱侧重吞吐一般如果团队是Java技术栈、需求集中在业务系统解耦、异步通知、跨系统事件ActiveMQ学习曲线平缓JMS标准写出来的代码还有迁移空间是稳的选择。如果是每秒百万级日志摄取、数据管道式处理ActiveMQ的吞吐不是它的强项流式处理平台更合适。如果业务里需要特别复杂的路由规则RabbitMQ的Exchange模型会更灵活。选型别只看性能先看团队能不能维护、业务模型合不合适。我在项目里见过为了“高端”强行引一套复杂组件结果半年没人会配置最后反而回退到ActiveMQ的情况。5.2 一个典型异步链路的参数估算用一个场景估算方便知道ActiveMQ的容量大致怎么算。假设订单系统下单后要异步通知短信服务、积分服务、库存服务峰值每秒下单500笔每笔产生一条JSON消息消息体平均2KB。目标是把消费延迟控制在5秒以内。第一步算每秒消息量500条每条2KB每秒1MB的数据量带宽和内存都不是瓶颈。第二步看消费能力假设单线程消费者处理一条消息平均20ms一个消费者每秒只能处理50条要撑住500条/秒的峰值至少需要10个消费者线程考虑实际CPU抢占、GC停顿按照0.7的利用率系数建议配14到15个消费者线程。第三步算队列容量5秒延迟对应队列里积压2500条消息每条约2KB总共5MB任何机器都扛得住。这个估算说明ActiveMQ处理这种量级绰绰有余瓶颈通常在消费端业务逻辑不在Broker端。这类估算到这里就够了不必做得很精细。真到瓶颈时再去调prefetch、消费者并发、批量确认方向清楚知道下一步动哪里。6. 学习路线与压箱底建议6.1 值得照做的学习顺序我见过不少人在ActiveMQ上学习路线很拧巴先啃activemq.xml各种高级配置用一堆特有API结果JMS最基础的模型反而没吃透。正确的顺序应该反过来。第一步是彻底吃透JMS核心模型ConnectionFactory、Connection、Session、Destination、Producer、Consumer这六个角色的关系和生命周期最好能不看文档画出调用链这比记住任何配置都值钱。第二步是用最简代码跑通Queue和Topic两种模型的收发把ACK、持久化、事务每一个点都实践一遍体会每种模式对消息生命周期的影响。第三步才是看ActiveMQ的特性failover、虚拟目的地、消息选择器、监控接口这些是使用中的加分项。第四步建议看Broker配置和数据目录结构理解持久化、内存、日志、管控台到这一步你对ActiveMQ就有比较完整的认知了。反过来学一上来陷入细节很容易三个月只记住一堆参数遇到问题还是不会排查。6.2 几个容易忽略的注意事项最后把我压箱底的经验一次写完。Session不是线程安全的。同一个Session里的生产者、消费者、监听器不要在多个线程同时用多线程需要各自创建Session但Connection可以共享。很多人写并发消费时共享了一个Session结果时不时报错就是这个原因。MessageProducer和MessageConsumer都可以复用不用每条消息都新建但连接和会话的关闭时序要小心。try-with-resources里先关生产者、消费者最后关Connection顺序反了会导致资源释放不彻底。管理控制台里看到的消息数量不等于实际即时状态。控制台有刷新延迟只适合人工观察别拿它做监控告警数据源监控要用JMX或统计接口。ObjectMessage能不用就不用。跨语言、跨版本、外部客户端的场景下序列化问题能缠死你统一转JSON字符串是更省心的路子。broker嵌入模式。做单元测试时可以在测试里启动一个嵌入式Broker代码里创建BrokerService并start不需要外部进程测试稳定性和速度都很理想这也是我调试时最顺手的一招。最后再分享一个细节我每次交付ActiveMQ相关项目都会留下一份速查笔记把连接URL、关键配置、排错路径、幂等策略写清楚。因为ActiveMQ本身不难难的是半年后再让人接手时所有坑都重新踩一遍。“学习一则搜”这个标题其实就是想说与其把经验留在脑子里不如整理成一份随查随用的手册下次遇到问题翻自己的笔记比重新翻文档快得多。