我第一次在生产环境里正儿八经用ActiveMQ是在一个订单系统里做状态同步。那之前我对JMS的理解一直停留在“有一组接口有个消息队列”的程度结果一上手就发现规范、实现、配置、可靠性这几件事每一个都能写出几千字的笔记。这篇内容就是我把“JMS, ActiveM 学习一则”这个标题补全成ActiveMQ之后整理出的完整学习记录适合刚接触消息中间件、准备在项目里引入MQ的开发者也适合那些已经在用、但还没系统捋过JMS概念的工程师。我会从JMS规范讲起用ActiveMQ 5.x做演示覆盖基础用法、消息可靠性和排障思路尽量做到看完能直接抄作业。1. JMS到底是什么先搞懂它要解决什么问题1.1 没有消息中间件时系统之间是怎么通信的很多项目一开始没有MQ服务之间就是简单的HTTP同步调用。下单接口里调积分服务调短信服务调库存服务每个调用都在同一个线程里串行执行。看起来逻辑清晰代码直接但问题会在流量上来之后集中爆发。首先是耦合。下单接口的成功依赖所有下游服务的成功。短信通道超时下单就超时积分服务重启下单就失败。为了不让一个次要的积分接口拖垮主流程大家开始加超时时间、加熔断器于是代码里塞满了try-catch和降级逻辑。核心业务代码和非核心逻辑混在一起改动一个需求要瞻前顾后。其次是流量冲击。日常订单量每秒几百数据库扛得住到了促销节点瞬时流量翻几十倍数据库连接池最先被打满接着下游服务集体雪崩。系统不是被一个请求打死的而是被一秒钟成千上万个请求压死的。这时候把“下单”和“后续处理”拆开中间架一个消息队列让下单请求先落库再把“订单已创建”这个事件丢到队列里积分、短信、物流各自去消费原本同步耦合的链路就变成了异步协作。这就是消息中间件存在的根本价值异步、解耦、削峰填谷。1.2 JMS规范和ActiveMQ的分工JMS全称是Java Message Service它是Java平台上的一套消息服务API标准接口都定义在javax.jms包里。它只规定“Java程序怎么发消息、怎么收消息”不规定底层怎么传输、怎么存储。你可以把JMS理解成Java世界里的JDBCJDBC定义了一堆数据库操作的接口底层具体连MySQL还是PostgreSQL由对应驱动实现JMS定义了一堆消息操作的接口底层具体用哪个消息中间件由对应实现类负责。ActiveMQ就是这个标准下的一个具体实现而且是开源阵营里非常老牌的一个。它实现了JMS 1.1的大部分规范同时兼容JMS 2.0的一些特性。与之类似的还有IBM MQ、OpenMQ等但ActiveMQ因为免费、轻量、部署简单一直是学习和中小团队落地的首选。还有个容易混淆的概念是AMQP。JMS是API层面的规范AMQP是协议层面的规范。简单说JMS规定的是“你代码里怎么调用”AMQP规定的是“消息在线路上长什么样”。ActiveMQ对两者都支持你用JMS API操作底层可以走ActiveMQ自己的OpenWire协议也可以走AMQP协议。学习时先抓住JMS API这条主线就够了。1.3 什么时候该上MQ什么时候别硬上消息中间件不是银弹引入它是要付代价的。代价包括多了一个必须保证高可用的基础设施、消息可能出现重复和乱序、排查问题多了一层链路。我个人判断的标准很简单两个服务之间同步调用能跑得稳稳当当流量也没大到把数据库打垮的程度就别折腾MQ。真正适合上MQ的场景有三个业务上需要立刻响应用户后续处理可以延后。比如下单成功返回积分和短信慢慢发。上游流量有明显的瞬时高峰需要队列缓冲。比如定时任务集中触发、秒杀流量。多个系统都要关注同一件事。比如订单状态变化订单系统、支付系统、风控系统都要知道那就广播一条消息。反过来如果业务要求强一致比如转账必须同步扣款成功才能返回那MQ就不适合做主链路它只能做辅助通知。记住这一点后面很多设计决策都不会走偏。2. 从零搭一个ActiveMQ准备环境、启动、发第一条消息2.1 版本选型与最小化启动ActiveMQ现在有两个大分支5.x系列的Classic版本和6.x系列的Artemis版本。Artemis是下一代引擎性能和架构更现代但很多老项目的配置和客户端依赖还是基于Classic的。如果是学习JMS概念建议先用5.x因为资料多、遇到的坑基本都有人踩过如果是新项目选型可以直接考虑Artemis它的集群和性能模型更清晰。我本地用的是5.16.x版本。下载解压后目录结构大致是bin、conf、data、lib、webapps。启动很简单cd apache-activemq-5.16.5/bin ./activemq start启动后看日志确认没有异常tail -f ../data/activemq.log看到类似“ActiveMQ Web Console available at http://127.0.0.1:8161”的日志就说明起来了。ActiveMQ默认开放两个端口61616给客户端连接用8161给Web控制台用。可以用netstat确认netstat -an | grep 61616提示Windows环境直接运行activemq.bat start即可其他步骤完全一样。2.2 控制台里手动发一条消息Web控制台是学习ActiveMQ最快的入口。浏览器打开http://127.0.0.1:8161默认账号密码都是admin。第一次登录强烈建议立刻改掉默认密码生产环境用默认口令跟裸奔没有区别。在控制台左侧找到Queues新建一个名为ORDER_QUEUE的队列。然后点进这个队列在Send区域填写消息内容点发送。这时候你再看队列的Message Count会显示有1条消息待消费。控制台提供了最直观的“消息从生产到堆积”的体验看完你就理解为什么监控MQ的指标里第一个就是队列深度。控制台适合做实验但实际项目里不可能靠人去点。真正的生产者和消费者都要用代码写而且生产者和消费者可能是完全不同的两个应用。2.3 用Java写最小生产者和消费者创建一个Maven工程引入ActiveMQ客户端依赖dependency groupIdorg.apache.activemq/groupId artifactIdactivemq-client/artifactId version5.16.5/version /dependency想省事也可以直接引入activemq-all里面打包了broker相关类学习阶段没什么差别。生产环境用activemq-client更干净。生产者代码import javax.jms.*; import org.apache.activemq.ActiveMQConnectionFactory; public class SimpleProducer { public static void main(String[] args) throws Exception { // 1. 连接工厂指向本机61616端口 ConnectionFactory factory new ActiveMQConnectionFactory(tcp://127.0.0.1:61616); // 2. 创建连接并启动 Connection connection factory.createConnection(); connection.start(); // 3. 创建会话非事务自动确认 Session session connection.createSession(false, Session.AUTO_ACKNOWLEDGE); // 4. 创建目的地和生产者 Queue queue session.createQueue(ORDER_QUEUE); MessageProducer producer session.createProducer(queue); // 5. 发送文本消息 TextMessage message session.createTextMessage(order-1001 created); producer.send(message); System.out.println(sent: message.getText()); // 6. 关闭资源 producer.close(); session.close(); connection.close(); } }消费者代码import javax.jms.*; import org.apache.activemq.ActiveMQConnectionFactory; public class SimpleConsumer { public static void main(String[] args) throws Exception { ConnectionFactory 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); // 创建消费者并同步接收消息 MessageConsumer consumer session.createConsumer(queue); Message message consumer.receive(3000); if (message instanceof TextMessage) { System.out.println(received: ((TextMessage) message).getText()); } consumer.close(); session.close(); connection.close(); } }先跑生产者再跑消费者。控制台里ORDER_QUEUE的消息数量会从1变为0。这个最小链路跑通之后再去看JMS的编程模型很多东西就能对上号了。3. 把JMS编程模型里的关键概念梳理清楚3.1 Connection和Session到底什么关系第一次写代码时我总觉得一个Connection就够了Session似乎可有可无。后来踩了并发问题的坑才明白这两个角色对应的是完全不同的资源层级。Connection是应用程序和消息中间件之间的物理连接。它负责建立网络连接、认证、启动底层的传输。Connection本身是重量级的一个应用通常只需要创建一两个所以它被设计成线程安全的。Session是建立在Connection之上的轻量级会话负责创建消息、生产者、消费者也负责管理消息的确认和事务。Session不能多个线程共享它是线程不安全的。这就相当于数据库里的Connection和Statement的关系Connection是一条通到数据库的网络链路Statement是这条链路上的一次具体操作。多线程环境里最简单的做法是每个线程用自己的Session。我在一个并发场景里遇到过“消费者偶尔丢消息”的问题排查到最后发现是多个线程共用了同一个Session去receive导致确认状态互相干扰。换成每线程一个Session之后问题消失。这个教训基本写进了我后来所有的JMS项目规范里。3.2 Queue和Topic排队办事与电台广播JMS里有两种消息模型刚开始特别容易混我列个表直接用起来对比项Queue点对点Topic发布订阅消息去向一条消息只会被一个消费者拿走一条消息会发给所有订阅者消费后状态被消费后消息移除非持久订阅离线就收不到是否支持持久订阅天然持久存储需额外开启持久订阅生活类比银行排队叫号办事电台广播谁听谁知道典型场景订单处理任务分发订单状态变更广播通知Queue的逻辑好理解类似任务分配消息发进队列多个消费者竞争谁抢到谁处理处理完这条消息就没有了。Topic就要多想一层生产者和消费者在时间上需要匹配。非持久订阅的Topic消费者必须保持在线才能收到消息生产者在消费者离线时发的那批消息消费者一个都看不到。如果希望Topic消息在消费者离线期间也不丢就要使用持久订阅DurableSubscription。持久订阅需要给消费者指定一个唯一的clientIdbroker会为这个订阅者保存消息等它上线后补发。这个机制生产环境中很常用比如某个服务A在夜间需要监听服务B发布的价格变更就需要持久订阅否则夜里漏掉的价格更新可能影响第二天结算。3.3 消息的五种形态为什么TextMessage最实用JMS定义了五种消息接口TextMessage字符串消息、MapMessage键值对消息、BytesMessage字节数组消息、StreamMessage原始类型流消息、ObjectMessage可序列化对象消息。实际项目里我推荐优先用TextMessage里面放JSON字符串。理由很简单可读性好日志里能直接看内容跨语言友好其他语言的消费者也能解析版本兼容压力小数据结构变化只是JSON字段增减。ObjectMessage看起来很爽直接塞一个Java对象就能传但它是个大坑。生产者和消费者的类版本必须完全匹配稍微升级一个字段就可能序列化失败。我见过不止一次因为ObjectMessage导致消费端反序列化异常最后所有团队都改成约定JSON格式。MapMessage和BytesMessage的应用场景更特定前者适合简单的键值对通知后者适合传输文件内容和二进制数据。学习阶段不用太纠结记住TextMessage是万金油就行。4. 生产环境必须认真对待的可靠性细节4.1 持久化消息到底存在哪里消息发送到ActiveMQ之后在没有被消费之前它需要有个地方暂存。默认情况下ActiveMQ把消息存储在KahaDB这是它内置的基于文件的存储引擎数据目录默认是data/kahadb。KahaDB的优势是不依赖数据库启动快、性能也不错适合大多数场景。为什么存储这么重要因为如果消息只存在内存里一旦broker进程崩溃、机器断电所有堆积的消息都会消失。开启持久化之后消息在标记为已消费之前都会写到磁盘重启后还能恢复。队列的消息默认就是持久化的但我们仍然可以在生产端显式指定消息的投递模式producer.send(message, DeliveryMode.PERSISTENT, 4, 0);其中第二个参数是DeliveryMode.PERSISTENT第三个是优先级最后是过期时间。这里“持久化”保证的是broker收到消息后的存储可靠性和前面说的持久订阅不是一回事。理解这一点能避免调试时到处翻文档。注意持久化也意味着每条消息都要落盘吞吐量会比纯内存低。追求极致吞吐的场景可以使用异步发送和批量刷盘但要接受故障时极小概率丢失最近一批消息。4.2 消息确认机制什么时候才算“消费成功”消费者收到消息后broker怎么知道这条消息可以删掉了这就靠确认机制。JMS里Session的确认模式决定了消息在什么时机被确认。AUTO_ACKNOWLEDGE是默认模式。同步receive()拿到消息返回时自动确认异步MessageListener则在onMessage()正常返回时自动确认。这个模式简单但如果你在onMessage里做了部分业务逻辑然后抛了异常消息并不会因为异常而进入重投流程可能就静默丢了。CLIENT_ACKNOWLEDGE模式则把确认的主动权交到业务代码手里。消息被receive()返回后broker不会自动确认必须手动调用message.acknowledge()。调用时机由你控制可以消费完立即确认也可以批量处理完多条后统一确认。我处理重要消息时偏好CLIENT_ACKNOWLEDGE尤其是在“先更新数据库再确认消息”的场景里。只要消息没确认broker就会在未来重投这就给业务提供了“数据库写成功了但还没确认”的兜底机会。代价是消息会有重复消费的可能所以业务消费逻辑必须做幂等。幂等不是技术选项是引入MQ之后的必修课。4.3 事务性Session要么一起提交要么一起回滚除了确认模式Session还有一个事务开关。创建Session时第一个参数传true就是事务性会话Session session connection.createSession(true, Session.SESSION_TRANSACTED);事务性会话里多个消息的发送和接收可以被绑定在同一个事务中。只有调用session.commit()之后事务里的操作才对其他消费者可见如果调用session.rollback()当前事务里发送的消息会撤销收到的消息会重回队列待投递。这个机制适合批量消费场景一次拉取多条消息全部处理成功后一次性确认避免一条一条确认带来的性能损耗。但要注意事务性Session管的是MQ本身的事务它和你的数据库事务是两套体系。如果你在同一个事务里先改数据库再发MQ消息数据库提交了MQ还没提交或者反过来都可能产生两边状态不一致的结果。想彻底解决“数据库和MQ一致性”问题需要本地消息表或分布式事务框架这已经超出JMS本身的范围。学习阶段先把Session事务和确认机制的区别记清楚知道什么场景用哪个不踩低级错误就够了。4.4 死信队列消息最终的去向不是所有消息都能被成功消费。消息可能因为业务逻辑异常被反复重投也可能在队列里等待太长超过过期时间最终被broker判定为“无法处理”。这些消息不会无限滞留在队列里而是会进入一个特殊的队列死信队列默认名为ActiveMQ.DLQ。死信队列很像现实中的“问题件处理中心”。正常消费不掉的异常消息集中放在这里由专门的后台任务去查看、补录、修复数据。生产环境里一定要监控死信队列的深度死信突然上涨通常说明有消费者异常或数据本身有问题。可以把死信队列按业务拆分比如每个业务队列对应一个自己的DLQ方便定位问题policyEntry queueORDER_QUEUE deadLetterStrategy individualDeadLetterStrategy queuePrefixDLQ./ /deadLetterStrategy /policyEntry这段配置放在activemq.xml的destinationPolicy里。配置后ORDER_QUEUE的死信会进入DLQ.ORDER_QUEUE而不是统一堆到ActiveMQ.DLQ里。有了这个细分排查问题时的效率高很多。5. 高可用和性能调优的粗浅经验5.1 单机换成主从先解决存储共享单机ActiveMQ永远是最脆弱的一环。进程崩溃可以有守护进程拉起但机器断电、磁盘损坏这种故障单机就是单点。生产环境至少要部署主从架构。ActiveMQ主从最简单的实现方式是共享存储主从Shared File System Master/Slave。核心思路是多个broker进程共享同一份持久化存储目录谁先拿到存储锁谁就是Master对外服务Master故障后备份broker抢到锁自动接管。这种架构要求存储目录必须是被所有broker节点共享的比如共享SAN磁盘或者分布式文件系统但不建议用NFS锁机制在高延迟网络下表现不稳定。配置起来也不复杂把persistenceAdapter里的dataDirectory指向同一个共享目录即可。理解了原理剩下的可以边用边调。5.2 生产者和消费者两侧的关键参数生产端的第一个参数是连接超时时间。默认连接超时较长网络异常时会拖慢主业务流程。我习惯在连接URL里显式指定ActiveMQConnectionFactory factory new ActiveMQConnectionFactory(tcp://127.0.0.1:61616?connectionTimeout3000);这样broker不可达时3秒内就能快速失败不会让业务线程长时间挂起。生产端的第二个点是批量发送。单条消息单条消息地send每条都要走网络往返吞吐自然上不去。使用事务性Session攒一批消息后commit吞吐能提升一个量级。代价是批量提交的消息有一定的累积延迟适合对实时性要求不高的场景。消费端最容易被忽视的参数是预取数量Prefetch。ActiveMQ默认会提前拉取一批消息缓存在客户端本地提高消费效率。默认queuePrefetch是1000条这个值在消费者处理速度快时很合适但消费者下游是慢数据库时1000条消息积压在本地一旦消费者崩溃重新投递和确认的处理就复杂了。遇到“消费者处理不过来内存却一直涨”的情况先检查预取值。可以在连接URL里调整tcp://127.0.0.1:61616?jms.prefetchPolicy.queuePrefetch100数值需要根据业务吞吐反复压测不要指望抄一个固定值就一劳永逸。5.3 用JMX和控制台盯住队列深度ActiveMQ内置JMX支持默认可以通过jconsole连接。我常用的监控指标有三个队列深度Pending Message Count、消费者数量Consumer Count、消息平均大小。队列深度是核心指标它直接反映生产消费速度是否匹配。更简单的方式是看Web控制台的Charts和Views页面能直观看到各队列的走势。但它只适合临时排查生产环境还是要配合监控系统做告警。建议对每个关键队列设置“队列深度超过阈值持续N分钟就告警”的规则这是发现消费者异常最有效的手段。如果发现消息消费不过来了排查看两点先看消费者进程是否存活并正常消费如果存活看是不是消费者线程数太少或者下游瓶颈导致单条消息处理过慢。队列深度只是表象真正的问题需要顺着消费链路往下挖。6. 实战排障踩坑记录速查表6.1 八个典型问题与处理方式下面这份速查表基本覆盖了我学习ActiveMQ过程中踩过的、见过别人踩的高频问题场景现象常见原因处理方式消费者收不到Topic消息生产发了消息消费端无反应非持久订阅消费者不在线改用持久订阅或确保消费者先启动消息重复消费业务被重复执行消费者处理慢导致broker重投消费逻辑做幂等消费者内存持续上涨堆内存占用飙升Prefetch值过大大量消息堆积在本地调小queuePrefetchObjectMessage反序列化失败消费端抛ClassCastException两端类版本不一致改成TextMessage传JSON连接被拒绝JMSException连接失败broker没启动或61616端口不通检查进程和防火墙消息突然全部进DLQ消费异常重投超限业务处理异常没有catch查异常日志修复后重投DLQ消息控制台登录不上密码错误默认密码被改过确认配置或人工重置消费很慢但堆里没消息消费者空转消费者数量太少或receive超时设置太短调整消费者线程数和超时时间6.2 从“收不到Topic消息”到理解订阅关系有个实际经历很能说明问题。某次调试先启动生产者往Topic发了十条消息过了两秒才启动消费者结果显示一条都收不到。当时第一反应是代码写错了反复检查没有发现异常。后来才反应过来非持久订阅的Topic消息发出去时没有消费者在线broker不会为谁保存这些消息自然全丢了。那之后我只要调试Topic相关逻辑都会先问自己一个问题这个Topic是希望离线的人在下次上线时还能收到还是只通知当前在线的人如果是前者就必须用持久订阅并且消费者要配置唯一的clientId和订阅名。理解订阅关系后再写代码错误率下降了很多。6.3 重复消费不可怕可怕的是不带幂等还有一次线上告警某个消费者进程重启后数据库里多了几十条重复的流水。原因是消费者从broker拉了一批消息处理到一半进程崩了消息因为没有确认而被重新投递重启后的消费者又处理了一遍。这可以说是MQ使用中最常见的坑。消息传输链路里有太多可能导致重复投递的环节网络闪断、消费者超时、broker主从切换。只靠“我相信消息不会重复”这种想法迟早吃大亏。解决办法不是让MQ保证消息不重复而是让业务逻辑自己具备幂等性。最简单的做法是给消息加一个业务唯一标识比如订单号、流水号消费时先查库判断是否处理过处理过就直接跳过。我个人在实际项目里的建议是凡是涉及金额、状态变更、资源扣减这类关键操作的消费者一律幂等。幂等逻辑放在消费者最前面成本很低但能挡住绝大部分重复消费事故。这套思路跟用不用ActiveMQ关系不大只要用了MQ就得默认“消息可能重复”否则后患无穷。
