1. 项目概述为什么我们绕不开JMS如果你在Java企业级应用里摸爬滚打过一段时间尤其是在处理不同系统、不同模块之间需要“说话”的场景那你大概率听过或者用过JMS。JMS全称Java Message Service翻译过来就是Java消息服务。听起来挺官方的对吧但说穿了它就是一套Java API标准专门用来让不同的Java应用组件能够以一种异步、可靠、解耦的方式进行通信。你可以把它想象成应用世界里的“邮政系统”发送方生产者把信件消息投递到邮局消息服务器至于信件什么时候、以什么方式送到接收方消费者手里发送方并不需要实时等待和关心。我见过太多项目一开始为了图省事系统A要调用系统B的功能直接就用HTTP接口同步调用了。这在业务简单、流量不大的时候没问题可一旦系统B挂了、响应慢了或者流量高峰来了系统A就会被直接拖垮整个调用链雪崩。后来大家学聪明了开始用消息队列比如ActiveMQ、RabbitMQ但怎么用呢各写各的代码风格五花八门维护起来头大。这时候JMS的价值就凸显出来了——它定义了一套统一的接口规范。不管你底层用的是ActiveMQ、IBM MQ还是其他实现了JMS标准的消息中间件你上层的业务代码写法几乎是一样的。这就意味着你今天用ActiveMQ明天想换成更牛的性能怪兽你的业务代码可能只需要改个连接工厂的配置核心逻辑动都不用动。这种“面向接口编程而非具体实现”的思想正是JMS设计的精髓也是它历经多年依然在企业级架构中占据重要地位的原因。所以这篇指南的目的不是给你罗列JMS那几十个接口的API文档那玩意儿官网都有。我想做的是结合我这些年踩过的坑、填过的洞带你从“知道JMS是什么”到“明白为什么用它”再到“上手就能写出健壮、高效的生产级代码”。我们会从最核心的两个消息模型聊起手把手过一遍发送和接收消息的每一个步骤深入那些容易出错的细节比如事务、确认模式最后再聊聊在实际项目中如何根据业务场景做出正确的技术选型。无论你是刚开始接触消息中间件的新手还是想系统梳理JMS知识的老兵相信都能从中找到你需要的东西。2. JMS核心模型与概念深度解析在真正动手写代码之前我们必须把JMS里几个最核心的“零件”和它们之间的“组装关系”搞清楚。这些东西就像乐高积木的基础模块理解透了后面搭建任何复杂的消息流都会得心应手。2.1 两种核心消息域点对点与发布/订阅JMS主要定义了两种消息传递的域Domain你可以理解为两种不同的通信模式它们解决了不同场景下的问题。点对点Point-to-Point, PTP模型想象一下银行叫号系统。一个取号机生产者产生号码消息并将其放入一个特定的队列Queue中。大厅里等待的多个窗口消费者都可以从这个队列里取号但一个号码只会被其中一个窗口消费掉消费后就从队列中移除。这就是点对点的精髓一条消息只有一个消费者能收到。队列天然具有负载均衡的能力多个消费者共同消费一个队列可以提高处理能力。同时队列还具有消息持久化的能力即使消费者暂时离线消息也会保存在队列中等待消费者上线后消费。这种模型非常适合任务分发、订单处理等“一个任务只需被处理一次”的场景。发布/订阅Publish/Subscribe, Pub/Sub模型这个更像我们订阅报纸或关注微信公众号。一个出版社生产者发布一期新的杂志消息到一个主题Topic。所有订阅了这个主题的读者消费者都会各自收到一份完整的杂志副本。这里的关键是一条消息会被所有当前活跃的订阅者消费。如果某个订阅者当时不在线非持久订阅那么它就会错过这期杂志。这种模型适用于广播通知、事件驱动架构中“一个事件需要通知多方”的场景比如用户注册成功需要同时发送邮件、初始化用户资料、发送欢迎短信等。注意这里有一个至关重要的区别。在P2P模型中消息的消费是竞争性的而在Pub/Sub中消息的消费是复制性的。理解这一点对你后续设计系统至关重要。比如你用Topic来发订单消息结果每个订阅的微服务都去扣一次库存那肯定就乱套了。2.2 JMS的核心组件与角色无论哪种模型都离不开下面这几个核心角色它们共同协作完成了消息的旅程。JMS Provider提供者这是消息系统的“基础设施”即实现了JMS规范的消息中间件产品本身比如我们常说的Apache ActiveMQ、RabbitMQ通过插件支持JMS、IBM MQ等。它负责消息的路由、传递、持久化、安全等底层脏活累活。JMS Client客户端使用JMS API来生产和消费消息的Java应用程序。也就是我们写的业务代码。Administered Objects受管对象这是为了解耦而引入的概念。像连接工厂ConnectionFactory和目的地Destination即Queue或Topic这些需要根据具体Provider配置的对象比如服务器地址、队列名称通常由管理员在JMS Provider上创建好然后通过JNDIJava命名和目录接口的方式提供给客户端代码查找和使用。这样客户端代码就不需要硬编码这些细节提高了可移植性。不过在Spring等现代框架中我们通常直接用配置类来定义这些Bean。ConnectionFactory连接工厂客户端用它来创建到JMS Provider的连接Connection。你可以把它看作一个连接池的入口里面封装了如何连接消息服务器的所有信息协议、地址、端口、用户名密码等。Connection连接代表客户端和JMS服务器之间的一个活动TCP连接。创建连接是一个相对昂贵的操作因此在实际应用中我们通常会使用连接池来管理。Session会话一个单线程的上下文用于生产和消费消息。它建立在Connection之上是实际进行消息操作创建生产者、消费者、消息的工厂。Session是JMS中事务和消息确认的基本单元这一点后面会详细展开。Message Producer消息生产者由Session创建用于向一个指定的目的地Destination发送消息的对象。Message Consumer消息消费者同样由Session创建用于从一个指定的目的地接收消息的对象。Message消息通信的载体。JMS定义了多种消息类型最常用的是TextMessage文本消息如JSON/XML字符串、MapMessage键值对、BytesMessage字节流和ObjectMessage序列化的Java对象但因其存在安全性和版本兼容性问题现在已不推荐使用。把这些组件串起来一个典型的流程是这样的客户端通过JNDI或配置获取ConnectionFactory- 用其创建Connection并启动 - 通过Connection创建Session- 通过Session创建指向某个Queue或Topic的Producer或Consumer- 进行消息的发送或接收 - 最后按顺序关闭所有资源Consumer, Producer, Session, Connection。3. 从零开始发送你的第一条JMS消息理论说再多不如动手跑一遍。我们以最常用的ActiveMQ作为JMS Provider来演示如何发送一条简单的文本消息。这里我会用最原生的JMS API这样你能看清每一个步骤后续再用Spring JMS之类的框架封装时你就能明白它帮你省了哪些事。3.1 环境准备与依赖引入首先你需要一个运行中的消息中间件。去Apache ActiveMQ官网下载最新版本解压后进入bin目录根据你的操作系统运行activemq start。默认的管理控制台地址是http://localhost:8161/admin默认账号密码是admin/admin。看到控制台说明服务启动成功了。接下来在你的Java项目Maven或Gradle中引入ActiveMQ的客户端依赖。如果你用Maven在pom.xml里添加dependency groupIdorg.apache.activemq/groupId artifactIdactivemq-client/artifactId version5.17.4/version !-- 请使用当前稳定版本 -- /dependency这个依赖会自动引入JMS API通常是javax.jms:javax.jms-api和ActiveMQ的具体实现。3.2 点对点模型消息发送实战假设我们有一个订单系统需要将新订单异步发送给库存系统进行扣减。我们创建一个名为ORDER.QUEUE的队列来处理这个任务。import org.apache.activemq.ActiveMQConnectionFactory; import javax.jms.*; public class OrderMessageSender { // ActiveMQ默认的Broker URL private static final String BROKER_URL tcp://localhost:61616; private static final String QUEUE_NAME ORDER.QUEUE; public static void main(String[] args) { Connection connection null; Session session null; MessageProducer producer null; try { // 1. 创建连接工厂 ConnectionFactory connectionFactory new ActiveMQConnectionFactory(BROKER_URL); // 2. 从工厂创建连接并启动注意启动后才能传输消息 connection connectionFactory.createConnection(); connection.start(); // 这一步非常关键经常被遗忘 // 3. 创建会话 (参数是否启用事务 消息确认模式) session connection.createSession(false, Session.AUTO_ACKNOWLEDGE); // 4. 创建目的地队列 Destination destination session.createQueue(QUEUE_NAME); // 5. 创建消息生产者并指定其发送的目的地 producer session.createProducer(destination); // 6. 创建一条文本消息 String orderJson {\orderId\: \1001\, \productId\: \P001\, \quantity\: 2}; TextMessage message session.createTextMessage(orderJson); // 7. 可选设置消息属性可用于消息过滤 message.setStringProperty(orderType, NORMAL); message.setIntProperty(priority, 5); // 8. 发送消息 producer.send(message); System.out.println(订单消息发送成功: orderJson); } catch (JMSException e) { e.printStackTrace(); } finally { // 9. 按顺序关闭资源 try { if (producer ! null) producer.close(); if (session ! null) session.close(); if (connection ! null) connection.close(); } catch (JMSException e) { e.printStackTrace(); } } } }代码逐行解析与避坑指南连接工厂我们直接实例化了ActiveMQ的实现类ActiveMQConnectionFactory。在生产环境中这个URL、用户名、密码通常会放在配置中心。connection.start()这是新手最容易掉进去的坑创建连接后必须调用start()方法连接才会真正开始工作才能传递消息。忘记调用会导致消费者收不到消息而且没有任何错误提示排查起来非常痛苦。创建会话的参数connection.createSession(false, Session.AUTO_ACKNOWLEDGE)。这里有两个至关重要的参数第一个参数transacted是否启用事务。false表示不启用会话事务。如果设为true则发送和接收的一系列操作可以作为一个原子单元提交或回滚。事务会话我们后面单独讲。第二个参数acknowledgeMode消息确认模式。Session.AUTO_ACKNOWLEDGE是最常用的表示当消息接收者成功从receive方法返回或消息监听器成功处理了消息后会话会自动确认消息。还有其他模式如CLIENT_ACKNOWLEDGE手动确认和DUPS_OK_ACKNOWLEDGE懒惰确认它们对消息的可靠性有不同影响是高级话题。消息属性除了消息体你还可以通过setXxxProperty方法设置一些自定义属性。这些属性不参与序列化消息体但可以被消费者用于消息选择器Message Selector实现基于属性的消息过滤功能非常强大。资源关闭一定要在finally块中按创建顺序的逆序关闭资源Producer - Session - Connection。虽然现代应用通常由框架管理生命周期但理解这一点有助于避免资源泄漏。运行这段代码如果控制台打印出发送成功的日志你就可以去ActiveMQ管理控制台的Queues页面看到ORDER.QUEUE队列里有一条消息在等待消费了。4. 消息的接收同步与异步两种模式消息发出去了总得有人来收。JMS提供了两种接收消息的方式同步阻塞接收和异步监听接收。它们适用于不同的业务场景。4.1 同步接收receive()方法同步接收很简单就是调用MessageConsumer.receive()方法这个方法会阻塞当前线程直到收到一条消息或等待超时。public class OrderMessageSyncReceiver { private static final String BROKER_URL tcp://localhost:61616; private static final String QUEUE_NAME ORDER.QUEUE; public static void main(String[] args) { Connection connection null; Session session null; MessageConsumer consumer null; try { ConnectionFactory factory new ActiveMQConnectionFactory(BROKER_URL); connection factory.createConnection(); connection.start(); // 创建非事务会话自动确认 session connection.createSession(false, Session.AUTO_ACKNOWLEDGE); Destination destination session.createQueue(QUEUE_NAME); // 创建消费者 consumer session.createConsumer(destination); System.out.println(等待接收订单消息...); // 同步接收等待最多10秒 Message message consumer.receive(10000); if (message ! null message instanceof TextMessage) { TextMessage textMessage (TextMessage) message; String orderInfo textMessage.getText(); String orderType textMessage.getStringProperty(orderType); System.out.println(收到订单消息类型: orderType , 内容: orderInfo); // 这里进行实际的业务处理比如扣减库存 // processOrder(orderInfo); } else { System.out.println(等待超时未收到消息。); } } catch (JMSException e) { e.printStackTrace(); } finally { // 关闭资源... } } }receive()方法可以传入一个超时时间毫秒receive(0)表示无限等待receive(1000)表示等待1秒。同步接收模式通常用于简单的测试、命令行工具或者需要严格顺序控制、一次只处理一条消息的场景。但在高并发的服务器应用中它会严重浪费线程资源因此生产环境几乎都使用异步监听模式。4.2 异步接收MessageListener接口异步模式是我们处理消息的“标准姿势”。你需要实现javax.jms.MessageListener接口并注册到消费者上。当消息到达时JMS Provider会自动调用你的监听器方法。public class OrderMessageAsyncReceiver implements MessageListener { private static final String BROKER_URL tcp://localhost:61616; private static final String QUEUE_NAME ORDER.QUEUE; public static void main(String[] args) throws JMSException, InterruptedException { ConnectionFactory factory new ActiveMQConnectionFactory(BROKER_URL); Connection connection factory.createConnection(); Session session connection.createSession(false, Session.AUTO_ACKNOWLEDGE); Destination destination session.createQueue(QUEUE_NAME); MessageConsumer consumer session.createConsumer(destination); // 创建监听器实例并注册 OrderMessageAsyncReceiver listener new OrderMessageAsyncReceiver(); consumer.setMessageListener(listener); // 启动连接开始监听 connection.start(); System.out.println(异步监听器已启动等待消息...); // 主线程等待防止程序退出 Thread.sleep(60000); // 监听1分钟 // 实际应用中这里可能是保持Web容器运行或使用CountDownLatch等机制 connection.close(); } Override public void onMessage(Message message) { try { if (message instanceof TextMessage) { TextMessage textMessage (TextMessage) message; String orderInfo textMessage.getText(); System.out.println(Thread.currentThread().getName() - 异步收到消息: orderInfo); // 模拟业务处理 Thread.sleep(500); // 处理耗时500ms System.out.println(订单处理完成: orderInfo); } } catch (JMSException | InterruptedException e) { e.printStackTrace(); // 重要在实际生产中这里需要根据业务决定是重试、记录死信还是其他补偿措施 } } }异步模式的核心优势与注意事项非阻塞与高并发主线程不会被阻塞可以继续处理其他任务。JMS Provider会使用内部的线程池来调用你的onMessage方法从而高效处理大量并发消息。线程模型onMessage方法在哪个线程中被调用是由JMS Provider决定的。你不能假设onMessage总是在同一个线程中执行。这意味着你需要确保你的监听器实现是线程安全的。异常处理onMessage方法中抛出的任何异常通常都会被JMS Provider捕获并记录但不会导致连接中断。然而对于AUTO_ACKNOWLEDGE模式消息可能在业务处理失败前就已经被确认了导致消息丢失。这是使用自动确认模式的一个重大风险点。消息确认时机在异步模式下消息的确认发生在onMessage方法成功返回之后对于AUTO_ACKNOWLEDGE。如果方法内抛出异常消息的确认行为取决于Provider的实现可能不会被确认从而触发重投递。实操心得在生产环境中我强烈建议为onMessage方法配置一个全局的try-catch并在catch块中根据业务重要性进行精细化处理。对于核心业务消息可以考虑使用CLIENT_ACKNOWLEDGE模式进行手动确认确保业务逻辑成功完成后再确认消息。5. 消息传递的可靠性保障事务与确认模式详解消息中间件号称“可靠”但这份可靠性需要开发者正确使用事务和确认模式才能兑现。这是JMS中最容易混淆也最关键的进阶知识。5.1 会话事务Transacted Session在创建会话时如果将第一个参数设为true你就创建了一个事务性会话。Session session connection.createSession(true, Session.SESSION_TRANSACTED);在这个会话内所有发送的消息和确认的消息对于消费者都不会立即生效而是被放入一个“待办列表”。直到你显式调用session.commit()这个“待办列表”里的所有操作才会被批量提交到JMS服务器。如果中间发生错误你可以调用session.rollback()这个会话内所有未提交的操作都会被撤销。发送端的事务示例Session session connection.createSession(true, Session.SESSION_TRANSACTED); MessageProducer producer session.createProducer(queue); // 发送多条消息 producer.send(session.createTextMessage(Message 1)); producer.send(session.createTextMessage(Message 2)); // 此时消息还在客户端缓存服务器未收到 // 模拟一个业务操作 boolean businessSuccess doSomeBusiness(); if (businessSuccess) { session.commit(); // 两条消息被原子性地发送到服务器 } else { session.rollback(); // 两条消息都被丢弃不会发送 }接收端的事务示例配合消费者Session session connection.createSession(true, Session.SESSION_TRANSACTED); MessageConsumer consumer session.createConsumer(queue); connection.start(); Message message consumer.receive(); // 处理消息... boolean processSuccess processMessage(message); if (processSuccess) { session.commit(); // 确认消息消费成功消息从队列移除 } else { session.rollback(); // 消息消费失败消息会重新放回队列通常会被重新投递 }事务会话的核心要点原子性commit时会话内所有操作发送和/或确认作为一个整体成功或失败。性能开销事务会带来额外的网络往返commit指令和服务器端锁开销性能低于非事务会话。适用场景适用于需要将多条消息的发送或者消息的消费与本地数据库操作保持原子性的场景。例如“扣减库存并发送扣减成功消息”必须同时成功或失败。5.2 消息确认模式Acknowledge Mode当会话为非事务transactedfalse时你需要通过确认模式来控制消息何时被确认为“已成功消费”。确认模式在创建会话时指定。Session.AUTO_ACKNOWLEDGE自动确认同步接收在receive()方法成功返回消息时会话自动确认该消息。异步接收在监听器的onMessage方法成功返回即未抛出异常时会话自动确认该消息。风险如果onMessage方法中业务处理失败例如数据库操作异常但方法本身正常返回了消息依然会被确认并移除导致消息丢失。这是最方便但最不安全的模式仅适用于消息处理非常轻量且允许丢失的场景。Session.CLIENT_ACKNOWLEDGE客户端手动确认消息的确认完全由客户端代码控制。消费者收到消息后必须调用message.acknowledge()方法来确认。调用acknowledge()会确认当前会话中所有已被接收但尚未确认的消息。这意味着如果你先处理了消息1然后收到消息2再确认消息2那么消息1也会被连带确认。优势你可以在业务逻辑确保成功如数据已入库后再确认消息避免丢失。示例session connection.createSession(false, Session.CLIENT_ACKNOWLEDGE); consumer session.createConsumer(queue); connection.start(); Message message consumer.receive(); try { // 处理业务逻辑 boolean success doBusiness(message); if (success) { message.acknowledge(); // 业务成功手动确认 System.out.println(消息已确认。); } else { // 业务失败不确认消息可能会被重投 System.out.println(业务处理失败消息未确认。); // 注意此时不应调用acknowledge会话关闭或连接关闭可能导致消息重投 } } catch (Exception e) { // 发生异常不确认 e.printStackTrace(); }Session.DUPS_OK_ACKNOWLEDGE重复消息优化确认这是一种“懒惰确认”模式。会话会延迟确认消息以减少会话开销提高吞吐量。代价是JMS Provider可能会认为消息未确认而重新投递即“重复投递”因此消费者必须能够处理重复消息业务逻辑需幂等。适用于可以容忍重复消息但对吞吐量要求极高的场景。如何选择一个简单的决策流需要与本地数据库操作保持强一致性 - 使用事务会话(transactedtrue)。不需要事务但要求每条消息必须在业务成功后确认且能接受稍低的吞吐 - 使用**CLIENT_ACKNOWLEDGE**。不需要事务业务处理简单快速且允许极低概率的消息丢失 - 使用**AUTO_ACKNOWLEDGE**。不需要事务追求极高吞吐且业务逻辑天然幂等或已做防重处理 - 可以考虑DUPS_OK_ACKNOWLEDGE。6. 生产环境中的高级特性与最佳实践当你的应用从Demo走向生产面对海量消息、复杂业务和故障常态时就需要掌握JMS的一些高级特性和经过验证的最佳实践。6.1 消息选择器Message Selector消息选择器允许消费者只接收那些匹配特定条件的消息。条件基于消息的**属性Property**和消息头Header进行过滤使用类似于SQL92条件表达式的语法。生产者设置属性TextMessage msg session.createTextMessage(重要订单); msg.setStringProperty(priority, HIGH); msg.setBooleanProperty(validated, true); msg.setIntProperty(amount, 10000); producer.send(msg);消费者使用选择器// 只接收 priority 为 HIGH 且 amount 大于 5000 的消息 String selector priority HIGH AND amount 5000; MessageConsumer consumer session.createConsumer(destination, selector);使用场景与限制场景分流处理。例如将高优先级订单路由到快速处理队列低优先级路由到慢速队列。或者让不同的消费者处理不同类型的消息。限制选择器是在JMS Provider服务器端进行过滤的。不匹配的消息根本不会传递给消费者这节省了网络带宽和客户端资源。但选择器表达式不能基于消息体Body内容进行过滤因为服务器端可能不会解析消息体。6.2 持久化消息 vs 非持久化消息消息的持久化决定了在JMS Provider重启或崩溃后消息是否会丢失。持久化消息DeliveryMode.PERSISTENT这是默认模式。发送时使用producer.setDeliveryMode(DeliveryMode.PERSISTENT)。消息会被保存到Provider的持久化存储如KahaDB、LevelDB、JDBC数据库中即使服务器重启消息也会恢复。用于保证消息不丢失。非持久化消息DeliveryMode.NON_PERSISTENT性能更高因为不需要磁盘IO。但服务器重启后消息会丢失。用于传输实时性要求高、允许丢失的数据如股票价格实时推送、在线游戏状态同步。// 发送持久化消息默认 producer.send(message); // 或显式设置 producer.setDeliveryMode(DeliveryMode.PERSISTENT); producer.send(message); // 发送非持久化消息 producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT); producer.send(message);选择建议对于订单、支付、库存扣减等核心业务消息必须使用持久化消息。对于日志收集、实时监控等辅助性、可再生的数据可以考虑非持久化以提升性能。6.3 消息过期、优先级与延迟投递消息过期Time To Live, TTL你可以设置消息的有效期。超过这个时间Provider会自动从目的地中删除该消息而不会投递给消费者。// 设置消息存活时间为5分钟300000毫秒 producer.setTimeToLive(300000); producer.send(message);这可以防止因消费者长时间离线导致队列中堆积“过时”的无用消息。消息优先级JMS支持0-9共10个优先级9为最高。优先级高的消息会被优先投递给消费者。但并非所有Provider都严格保证优先级顺序它只是一个提示。message.setJMSPriority(9); producer.send(message);延迟投递一些高级的JMS Provider如ActiveMQ支持延迟消息。消息不会立即进入队列而是在指定的延迟时间后才可供消费。// ActiveMQ特有的属性设置延迟60秒 long delay 60 * 1000; message.setLongProperty(AMQ_SCHEDULED_DELAY, delay); producer.send(message);这常用于实现定时任务如“30分钟后检查订单是否未支付若未支付则自动取消”。6.4 生产环境最佳实践清单连接、会话、生产者和消费者的池化频繁创建和销毁这些对象开销巨大。务必使用连接池如org.messaginghub:pooled-jms或依赖框架如Spring JMS的CachingConnectionFactory、JmsTemplate来管理它们。妥善处理异常与重试在MessageListener的onMessage中必须进行完整的异常捕获。区分业务异常如库存不足和系统异常如网络超时、数据库连接失败。对于系统异常应实现重试机制。可以将处理失败的消息发送到一个“重试队列”并设置递增的延迟时间或者使用支持重试的框架如Spring Retry。死信队列Dead Letter Queue, DLQ当一条消息因为某些原因如格式错误、业务逻辑始终无法处理、超过重试次数无法被成功消费时它应该被转移到DLQ。这可以防止“毒药消息”阻塞正常队列。ActiveMQ等Provider通常有默认的DLQ策略ActiveMQ.DLQ但你最好为关键业务队列配置专属的DLQ并设置监控告警。消费者数量与并发控制不要盲目增加消费者数量。对于顺序敏感的消息多个消费者会破坏顺序。同时消费者数量受限于会话和连接资源。需要根据消息到达速率和单条消息处理耗时动态调整消费者数量。监控与告警监控队列深度积压消息数、消费者数量、消息出入速率。队列深度持续增长是危险的信号可能意味着消费者处理能力不足或出现了故障。消息体设计强烈建议使用TextMessage并以JSON或XML格式传递数据。避免使用ObjectMessage因为它强耦合于Java语言和特定的类版本。存在反序列化安全漏洞如Java反序列化攻击。不利于其他语言系统的交互。JSON是跨语言、跨平台的通用选择。7. 常见问题排查与性能调优实录即使按照最佳实践来在实际运行中还是会遇到各种稀奇古怪的问题。下面是我在运维消息系统时经常遇到的几个典型场景和排查思路。7.1 消息堆积消费者不消费这是最常见的问题。表现就是管理控制台上看到队列里的Number Of Pending Messages待处理消息数不断上涨而Number Of Consumers消费者数可能为0或不变。排查步骤检查消费者应用状态首先确认消费此队列的应用程序是否正在运行日志是否有错误。网络是否通畅检查连接与启动确认消费者代码中connection.start()是否被调用。这是我见过新手最常犯的错误没有之一。检查消息选择器如果消费者设置了过于严格或错误的选择器Selector可能导致没有消息能匹配从而表现出“不消费”的假象。可以临时去掉选择器测试。检查确认模式与异常如果是AUTO_ACKNOWLEDGE模式检查onMessage方法是否抛出了未被捕获的异常在异步模式下一个持续抛出异常的监听器可能会导致会话被Provider标记为有问题从而停止向其投递消息。查看消费者列表在ActiveMQ控制台的队列详情页查看Subscribers部分确认是否有活跃的消费者连接。如果没有回到步骤1。7.2 消息被重复消费在非事务、CLIENT_ACKNOWLEDGE或网络不稳定的情况下消息可能被重复投递。原因与对策原因1消息确认后网络故障。消费者确认了消息但确认信息在网络传输中丢失Provider认为消息未确认于是重新投递。原因2消费者处理时间过长。超过了Provider配置的“重新投递延迟”时间Provider认为消费者挂了将消息重新投递给其他消费者。对策业务逻辑幂等。这是解决重复消费问题的根本方法。无论同一条消息收到多少次处理结果都应该是一样的。常见做法数据库唯一约束利用订单号、流水号等业务唯一键。状态机只有处于特定状态的消息才被处理。例如订单状态从“待支付”到“已支付”只能发生一次。分布式锁/令牌表在处理前先获取一个基于消息ID的锁。调整Provider配置适当增加ActiveMQ的重新投递策略redeliveryPolicy中的最大重试次数和初始重试延迟给消费者更长的处理时间。7.3 性能瓶颈分析与调优当消息吞吐量达不到预期时可以从以下几个层面排查1. 网络与IO层面使用NIO传输ActiveMQ默认使用TCP传输。对于大量小消息可以尝试启用NIO传输器tcp://localhost:61616?transport.useNiotrue它能用更少的线程处理更多连接。调整TCP缓冲区适当增加tcpBufferSize和ioBufferSize。2. 持久化层面对持久化消息影响巨大选择合适的持久化适配器ActiveMQ默认的KahaDB适用于大部分场景。如果追求极致性能且消息可丢失可临时切到非持久化。如果要求高可靠且数据量巨大可评估LevelDB或JDBC配合高性能数据库。关闭日志同步在KahaDB配置中可以设置enableJournalDiskSyncsfalse。这会提高性能但在系统崩溃时可能丢失最后几条未刷盘的消息通常可接受。3. 客户端层面使用异步发送ActiveMQConnectionFactory可以设置发送模式为异步setUseAsyncSend(true)。这能极大提升发送端的吞吐但发送者无法立即知道消息是否成功到达服务器。调整预取限制Prefetch Limit消费者会预先从服务器拉取一批消息缓存到本地。如果这个值太大默认1000当某个消费者处理慢时大量消息会堆积在该消费者本地导致其他消费者空闲。可以针对慢消费者调小此值如设置为1实现更公平的负载均衡。// 在连接URI中设置所有消费者的预取限制为10 String url tcp://localhost:61616?jms.prefetchPolicy.all10; // 或单独设置队列的预取限制 String url tcp://localhost:61616?jms.prefetchPolicy.queuePrefetch10;会话和连接的池化再次强调这是提升客户端性能的基础。调优是一个权衡的过程。提高性能往往意味着牺牲一定的可靠性或增加资源消耗。你需要根据业务的实际SLA服务等级协议来找到平衡点。我的经验是先从客户端代码和配置优化入手如预取限制、异步发送再考虑服务器端配置最后才考虑升级硬件。同时完善的监控是调优的眼睛没有监控的调优就是盲人摸象。