我在做订单系统的时候遇到过最头疼的一种问题下游库存服务偶发性变慢一次同步的HTTP调用等上三秒五秒订单线程全被拖死数据库连接池也被打满。后来把扣库存、发物流通知这些环节从同步调用改成了消息异步流转整个系统的抖动一下子小了。这也是我为什么会专门写这篇Spring开发系列教程(29)——集成JMS。JMS就是Java平台上的消息服务标准Spring对它做了非常完善的封装让你不用直接碰那些繁琐的原生API几行配置就能把消息中间件接入项目。这篇文章我会从为什么需要它、环境怎么搭、生产怎么避坑一路讲下来适合那些已经会用Spring Boot、但还没系统接触过消息队列的开发者。1. 为什么要把JMS请进项目从“同步等待”到“异步流转”1.1 同步调用的痛点你一定见过先说说没有消息队列时系统是怎么工作的。假设你的订单服务下单成功后需要调用库存服务扣减库存再调用物流服务创建运单。这三个操作如果用HTTP接口同步调用就形成了一条串行链路订单线程发出库存请求后必须一直等着库存服务返回结果才能继续调物流服务。问题出在链路中任何一个环节变慢或者干脆超时。库存服务数据库慢查询了订单线程就卡在等待上线程一多Tomcat线程池耗尽新的订单请求全部排队跟着数据库连接池也被这些卡住的请求占满整个服务就像多米诺骨牌一样倒下去。我在实际项目里看到过非常典型的连锁故障上游一个慢接口把下游几个服务的连接池全部打爆。这种架构还有一个更隐形的问题——耦合。订单服务的代码里如果直接写死了库存服务的URL那服务地址一变订单服务就得重新发布库存服务如果今天要求加字段订单服务也要跟着改。两个服务的发布节奏被绑在一起时间久了谁都不想动这块代码。1.2 JMS怎么改变这一切JMS解决这个问题的思路很简单也很彻底生产者把消息发到中间件中间件存下来消费者按自己的节奏去取。订单服务下单成功后往“扣库存”这个队列里丢一条消息然后立刻就往下走了不等待任何人的响应。库存服务什么时候拿这条消息、什么时候处理完都由自己决定。这样一来原来最要命的“等待时间”消失了。订单线程发送消息的动作非常快毫秒级完成库存服务就算处理得慢它在自己的消费者线程里慢慢处理也拖不垮订单服务。这就是异步解耦。除此之外JMS还能顺带解决削峰问题。比如秒杀场景瞬间进来一万个请求数据库根本扛不住那就先让请求把消息写进队列后端消费者批量处理以自己能承受的速度把这一万条任务消化掉。打个比方同步调用就是你站在餐厅柜台前盯着厨师把菜做完端给你JMS消息队列是你点完单拿个号就走后厨做好了再叫你。对餐厅来说柜台前不排长队了对客人来说干等的时间也消掉了。1.3 JMS和Kafka、RocketMQ这些新面孔是什么关系很多刚接触消息队列的读者会问现在不是流行Kafka和RocketMQ吗学JMS还有意义吗我的看法是要看系统所处的技术环境。如果你从零开始做一套全新的互联网高并发系统Kafka或者RocketMQ确实是热门选项这没问题。但在大量成熟的Java企业级系统里JMS规范下的ActiveMQ、老牌的MQ产品依然承担着核心的消息流转职责。对比项JMS以ActiveMQ为例KafkaRocketMQ定位Java标准消息API分布式流处理平台分布式消息中间件消息模型点对点队列、发布订阅分区日志模型队列/发布订阅协议支持OpenWire、STOMP、MQTT等自定义TCP协议自定义Remoting协议使用难度较简单Spring封装完善中等偏大数据场景中等功能丰富典型场景企业系统异步解耦、系统间集成日志采集、流式计算业务消息、事务消息所以学JMS本质上是在学一套很通用的“异步消息”思维掌握了这套思维你再去碰Kafka、RocketMQ会发现很多概念是相通的队列、主题、消费者组、消息确认。Spring的JMS支持本身也很成熟加上Spring Boot的自动配置用起来远比想象中轻松。2. 搭建最简可运行环境五个配置项把消息中间件接进Spring2.1 依赖选择Boot自动配置省掉一半功夫我以ActiveMQ为例因为它是纯Java实现的开源消息中间件非常适合学习和功能验证。如果你用的是别的兼容JMS规范的产品核心配置思路基本一致。在Spring Boot项目里用ActiveMQ最简单的方式是引入对应的Starterdependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-activemq/artifactId /dependency如果工程里没有用Boot只有原生的Spring框架那就需要自己引入Spring的JMS模块和ActiveMQ客户端dependency groupIdorg.springframework/groupId artifactIdspring-jms/artifactId /dependency dependency groupIdorg.apache.activemq/groupId artifactIdactivemq-client/artifactId /dependency引入Starter之后Spring Boot的自动配置会帮你生成连接工厂、JmsTemplate这些核心Bean。这在项目初期的体验相当舒服——你甚至不必知道细节就能先把一条消息发出去。但要注意自动配置只是“帮你做了默认选择”不代表不需要了解底层生产环境里那些连接池、监听器并发参数后面多半还是要手动调整的。2.2 核心配置项详解如果使用内嵌模式的ActiveMQBroker和你的应用跑在同一个JVM里配置极其简单spring: activemq: broker-url: vm://embedded-broker?brokerPersistentfalse user: admin password: admin如果是连接独立的ActiveMQ服务器则把broker-url改成TCP地址spring: activemq: broker-url: tcp://192.168.1.100:61616 user: admin password: admin这里有几个值得注意的点。第一brokerPersistentfalse表示消息不持久化Broker重启消息就丢只适合本地开发调试生产环境一定要改成持久化模式。第二用户名密码默认是admin/admin生产环境务必改掉。第三Spring Boot的默认配置里连接工厂是CachingConnectionFactory它会缓存Session、Connection和Producer避免每次收发消息都重新创建连接这对性能很重要。不过在实际项目里我通常会在配置类中显式声明一个PooledConnectionFactory因为ActiveMQ官方推荐用连接池来提升高并发下的吞吐量。具体做法先不急下面会详细说。2.3 手动配置连接工厂的正确姿势当你需要控制连接池大小、设置重连策略时Boot的自动配置就不够用了。这时候我会在配置类里重写ConnectionFactory的Bean定义。一个我实践下来最好用的配置是这样的用的是Activemq连接池Configuration public class JmsConfig { Bean public ConnectionFactory connectionFactory() { ActiveMQConnectionFactory amq new ActiveMQConnectionFactory(); amq.setBrokerURL(tcp://192.168.1.100:61616); amq.setUserName(admin); amq.setPassword(admin); PooledConnectionFactory pooledFactory new PooledConnectionFactory(); pooledFactory.setConnectionFactory(amq); pooledFactory.setMaxConnections(10); pooledFactory.setMaximumActiveSessionPerConnection(20); pooledFactory.setIdleTimeout(300000); return pooledFactory; } Bean public JmsTemplate jmsTemplate(ConnectionFactory connectionFactory) { JmsTemplate jmsTemplate new JmsTemplate(); jmsTemplate.setConnectionFactory(connectionFactory); jmsTemplate.setDeliveryPersistent(true); return jmsTemplate; } }这里为什么要套两层工厂ActiveMQConnectionFactory负责和Broker建立底层连接而PooledConnectionFactory则把这些连接缓存起来复用。我见过不少初学JMS的人把这两层搞混以为只配置一个ActiveMQConnectionFactory就够了。在高并发场景下如果没有连接池每次发消息都新建物理连接TCP握手和认证的开销会直接把性能拖垮。还有一个容易被忽略的参数deliveryPersistenttrue。JmsTemplate默认发消息是持久化的也就是说Broker收到消息后会落盘。如果不小心把它设成了false消息在Broker重启后会丢这在生产环境属于重大事故级别的配置。3. 收发消息的核心链路JmsTemplate、Destination与消息转换器3.1 JmsTemplateSpring封装给开发者的消息发送助手JmsTemplate是Spring JMS给开发者提供的核心工具类它的职责非常像RestTemplate之于HTTP调用封装底层的连接获取、Session创建、消息发送这些琐碎细节让你只关心业务内容。最常用的发送方式有两种。第一种是直接指定目的地名称Autowired private JmsTemplate jmsTemplate; public void sendDemo() { jmsTemplate.convertAndSend(order.queue, hello jms); }第二种是传入一个Destination对象Autowired private Destination orderQueue; public void sendWithDestination() { jmsTemplate.convertAndSend(orderQueue, hello jms); }convertAndSend这个方法的妙处在于它会自动把第二个参数普通Java对象转换成Message。默认情况下字符串会被包装成TextMessage字节数组会被包装成BytesMessage可序列化的Java对象会被包装成ObjectMessage。这个转换动作由MessageConverter完成默认的SimpleMessageConverter规则足够应对80%的场景。如果想发送一个Map或者自定义消息头、设置延迟时间就得手动构造Message了public void sendCustomMessage(String orderId) { jmsTemplate.send(order.queue, session - { TextMessage message session.createTextMessage(orderId); message.setStringProperty(orderType, NORMAL); message.setIntProperty(retryCount, 0); return message; }); }这里我用了lambda表达式配合send(String, MessageCreator)的重载方法。MessageCreator接口是个函数式接口我们在里面创建并返回一个Message。这种方式相比convertAndSend能做的事情更细比如你说要给某个消息设置10分钟延迟就只能在Message里放一个ScheduledMessage.AMQ_SCHEDULED_DELAY属性这个只有手动构造消息才能做到。3.2 Destination队列和主题两条不同的消息航道JMS规范定义了两种消息模型对应两种Destination模型Destination类型消息投递方式适用场景点对点Queue队列一条消息只被一个消费者消费任务分发、异步处理发布订阅Topic主题一条消息被所有订阅者消费广播通知、事件分发点对点模型在日常生活中对应一个队列式的快递柜你放进一件快递只会有一个人取走它。即便同时有十个消费者线程盯着这个柜子快递最终也只会被其中一个拿走。发布订阅模型对应的是杂志订阅出版社把刊物一印所有订阅了这本杂志的人都会收到一份。在Spring配置里我会为常用的队列和主题创建专门的BeanConfiguration public class DestinationConfig { Bean public Queue orderQueue() { return new ActiveMQQueue(order.queue); } Bean public Topic orderEventTopic() { return new ActiveMQTopic(order.event.topic); } }这样在注入Destination时代码的意图非常清楚。而ActiveMQ还有一个虚拟主题的概念可以结合队列和主题的优势生产者只往主题发消费者用“队列风格”的地址去消费保证每个消费组只有一台机器消费。这个我后续在讲集群消费时会再展开。3.3 MessageConverter消息格式的翻译官默认的SimpleMessageConverter处理字符串、字节数组、可序列化对象都没问题但有一种情况会踩坑往convertAndSend里传一个没有实现Serializable接口的对象运行时会直接抛异常。我早期在Spring集成JMS时就在这上面栽过跟头——传了个普通DTO结果消费者端一直收不到消息控制台却没有任何报错查了半天才发现是转换器在背后默默抛了异常因为JmsTemplate默认把异常吞掉了为什么吞后面会讲。在实际开发中消息体通常以JSON格式传输比较通用因为消费端可能是Java之外的语言。这时候我会自定义一个MessageConverterComponent public class JacksonJmsMessageConverter implements MessageConverter { private final ObjectMapper objectMapper new ObjectMapper(); Override public Message toMessage(Object object, Session session) throws JMSException, MessageConversionException { try { String json objectMapper.writeValueAsString(object); return session.createTextMessage(json); } catch (JsonProcessingException e) { throw new MessageConversionException(对象转JSON失败, e); } } Override public Object fromMessage(Message message) throws JMSException, MessageConversionException { if (message instanceof TextMessage) { try { return objectMapper.readValue(((TextMessage) message).getText(), Object.class); } catch (IOException e) { throw new MessageConversionException(JSON转对象失败, e); } } return null; } }然后把这个转换器塞进JmsTemplate和监听器容器里。这样做的好处非常直接消息体是纯JSON文本跨语言、跨系统的兼容性极好而且调试时通过管理界面看到的Queue里的数据就是可读的JSON而不是一串Java对象序列化后的二进制乱码。3.4 默认异常吞噬是坑也是设计我第一次用JmsTemplate时发现发送消息如果失败应用居然不报错。这个“静默失败”的行为一度让我很困惑。后来查了源码才明白JmsTemplate默认把所有异常都包在JmsException里抛出但如果你在回调里处理了异常或者某些异常被内部吞掉表面上就看不到错误。我不建议在生产环境长期依赖默认行为。把jmsTemplate.setPriority这类发送优先级暂且不说更关键的是设置setExplicitQosEnabled(true)并且打开日志logging: level: org.springframework.jms: DEBUG这样每次发送是否成功、用了多少时间日志里能看得清清楚楚。排查消息丢失这类问题的时候开启DEBUG日志往往是第一步。4. 消息监听与消费注解驱动的完整姿势4.1 JmsListener最优雅的消费端写法消息发送用JmsTemplate消息接收则靠监听器。Spring JMS提供了JmsListener注解你只需要在一个Bean方法上标上它容器就会自动为该目的地创建消费者的监听线程Component public class OrderMessageConsumer { JmsListener(destination order.queue, concurrency 1-3) public void onOrderMessage(TextMessage message) { String orderId message.getText(); System.out.println(收到扣库存消息 orderId); // 业务处理... } }关于参数类型这里要特别说明监听方法可以直接接收TextMessage、MapMessage等JMS原生类型也可以接收普通Java对象——但前提是注册了对应的MessageConverter或者消息本身就是通过convertAndSend发送的可序列化对象。如果你什么都没配置直接写一个接收自定义DTO的方法启动时不会报错但一旦收到消息转换阶段就会出问题。我最推荐的做法是配好自定义MessageConverter后直接让监听方法接收自定义DTO业务代码里就不用关心消息类型了。4.2 监听器容器消费的幕后管家JmsListener之所以能生效是因为背后有一个MessageListenerContainer在默默工作。Spring Boot默认给你创建的是DefaultJmsListenerContainerFactory它是基于Spring的DefaultMessageListenerContainer实现的支持事务、并发调整、异常处理等高级特性。在配置类里可以自定义这个工厂Configuration public class JmsListenerConfig { Bean public JmsListenerContainerFactory jmsListenerContainerFactory( ConnectionFactory connectionFactory, MessageConverter messageConverter) { DefaultJmsListenerContainerFactory factory new DefaultJmsListenerContainerFactory(); factory.setConnectionFactory(connectionFactory); factory.setMessageConverter(messageConverter); factory.setSessionTransacted(true); factory.setConcurrency(2-5); factory.setAutoStartup(true); return factory; } }这里sessionTransacted(true)特别关键它表示消费消息和业务操作在同一个事务中完成。为什么重要因为JMS的消费模型是“消费者拿到消息处理完后再确认”。如果消费到一半抛异常事务回滚这条消息就还在队列里之后会被重新投递。有了这个配置就能避免“消息已经处理了但没来得及确认结果重复消费”的经典问题。当然事务只对单条消息的处理过程有效它不能覆盖跨数据库、跨消息的分布式事务——那是另一个层面的大问题。concurrency参数控制了消费者并发范围。比如1-3表示最小1个、最多3个消费线程。数字越大并发处理能力越强但也要考虑下游资源和数据库连接。不建议一味调高我看过有人把并发调到20结果数据库连接池先被占满了消息消费反而更慢。这个参数要根据实际压测结果来调。4.3 同时消费队列和主题的两种方式JmsListener默认既能监听队列也能监听主题。如果监听的是Topic需要注意一个概念持久订阅和非持久订阅。非持久订阅的消费者宕机期间发布的消息就再也收不到了持久订阅则会让Broker把消息存着等消费者恢复后再投递。ActiveMQ里使用JmsListener监听主题时默认是非持久订阅。如果需要持久订阅要把订阅者的clientId配置好JmsListener( destination order.event.topic, subscription orderEventSubscriber, containerFactory jmsTopicListenerContainerFactory ) public void onOrderEvent(String eventJson) { // 处理订单事件 }我个人的习惯是能用队列解决的尽量用队列主题一般用在“多个不同系统都要收到同一类事件”的场景比如用户注册以后积分系统、消息推送系统、数据分析系统都要各自处理这个事件这时候一个主题就能给所有订阅者广播。而如果只是同一个系统内部的任务分发队列更可靠消息不会被重复消费。5. 生产环境躲不开的坑重连、幂等与消息堆积5.1 连接断开之后Broker重启与生产者感知生产环境里ActiveMQ偶尔会重启维护或者网络发生抖动。如果连接工厂没配好生产者在连接断开后可能长时间无法恢复。ConnectionFactory里有一个关键参数叫failover协议ActiveMQ把它封装成了transport方式ActiveMQConnectionFactory amq new ActiveMQConnectionFactory(); amq.setBrokerURL(failover:(tcp://192.168.1.100:61616)?maxReconnectAttempts5);failover:表示当主连接断开时客户端会自动重连并且重连期间发送的消息会缓存下来等连接恢复后再发出去。不配这个协议Broker一重启生产者端可能直接抛javax.jms.ConnectionClosedException消息发送直接失败。不过要注意failover把发送动作和实际投递解耦了也就是说消息并不会在连接断开时立刻报错。这有好有坏好处是发送接口更稳定坏处是你无法感知消息是不是真的已经发给Broker了。在极端情况下如果重连时间过长生产者内存里堆积的待发消息也会成为隐患。我一般会监控生产者的队列积压数量并设置合理的内存限制。5.2 消息幂等重复消费是常态不是意外只要用了消息系统就要接受一个现实消费者可能重复收到同一条消息。原因很多最常见的两个一是消费者处理完消息、但还没来得及向Broker确认时发生宕机Broker就会把这条消息重新投递二是在sessionTransacted(true)的事务下事务提交失败消息也会被重新处理。既然重复不可避免业务逻辑就必须做到幂等。幂等这个词听起来高大上意思其实很简单不管操作执行几次结果都和只执行一次相同。比如扣库存这个操作如果扣减逻辑里加上“订单状态为待处理时才允许扣减”那同一条消息进来两次第二次会因为状态已经变更而直接忽略。我常用的幂等方案有三种方案核心思路适用场景业务状态判断处理前查订单状态已处理则跳过订单类、支付类强业务场景唯一键兜底数据库唯一索引或Redis setNX通用场景防重复插入消息ID记录消费前检查已记录的消息ID无业务状态可依据时第三种方案我再用一段伪代码说明一下很实用JmsListener(destination order.queue) public void onMessage(Message message) { String msgId message.getJMSMessageID(); // 利用Redis的SETNX只有第一次能设置成功 Boolean first stringRedisTemplate.opsForValue() .setIfAbsent(msg:consumed: msgId, 1, Duration.ofHours(24)); if (first null || !first) { // 已经处理过这条消息直接返回 return; } // 真正执行业务逻辑 processOrder(message); }这套逻辑实现简单而且Redis的原子性足够保证并发场景下的幂等。唯一需要注意的是消息ID在JMS里并不是全局唯一不同连接生成的ID可能重复所以更稳妥的做法是自己在消息头里放一个业务唯一键比如orderUUID用它去做幂等键。5.3 消息堆积消费能力跟不上生产速度消息队列最健康的状态是生产者发一条消费者立刻取一条队列基本为空。一旦出现消费者消费速度长期小于生产速度队列里的消息就会越堆越多这就是消息堆积。在ActiveMQ的管理界面能看到队列的EnqueueCount和DequeueCount两者差额就是积压数量。消息堆积的根因通常有三个消费者吞吐量不足并发线程数设太低或者消费者里业务逻辑太复杂比如每消费一条都去做一次全表扫描。下游资源瓶颈消费者本身没问题但它的数据库连接、外部接口跟不上。消息处理失败反复重投某条消息处理总是失败导致它被反复投递从而拖慢了整个队列。针对不同根因处理方式也不一样。第一类问题调整JmsListener的concurrency参数或者优化消费者里的业务逻辑。第二类问题瓶颈在下游横向扩容消费者实例或者提升数据库能力才有用。第三类问题要在监听器里做重试次数判断超过一定次数的消息直接移到“死信队列”JmsListener(destination order.queue) public void onMessage(TextMessage message) { int retryCount message.getIntProperty(retryCount); if (retryCount 5) { jmsTemplate.send(order.queue.dead, session - session.createTextMessage(message.getText())); return; } try { process(message); } catch (Exception e) { // 重试次数1重新发送回队列 message.setIntProperty(retryCount, retryCount 1); jmsTemplate.convertAndSend(order.queue, message.getText()); throw e; // 抛出异常让Spring JMS认为消费失败不再确认 } }这个“先送死信队列再抛异常”的写法是我在生产里验证过的可靠方案。它既保证了异常消息不会无限重试拖垮队列又保留了完整的数据供人工或者补偿任务处理。5.4 监控与告警消息积压的“烟火报警器”做了上面那些配置也还不够。我始终认为可靠的系统不是靠“眼睛盯”盯出来的而是靠监控盯出来的。消息队列也是一样。部署一套基础监控重点盯三个指标队列积压数量超过阈值立刻告警。消费延迟时间生产者发送时间和消费者接收时间的差值能反映积压的严重程度。死信队列变化正常情况下死信队列消息很少一旦激增说明有大量消费失败。ActiveMQ自带Web控制台默认端口8161可以看到队列积压情况。更工程化的做法是把积压数据指标暴露给外部监控系统可以用JMX或者Prometheus的JMX exporter配好告警规则。我在项目里就吃过一次亏那天凌晨一个下游接口挂掉了消费者全部失败重试死信队列在一小时内积了十几万条消息第二天早上才被发现。从那以后我把消息积压告警的阈值设成了“超过500条立刻告警”。一个好的监控能让你在用户发现问题之前先发现系统问题。6. 一点个人的实战体会6.1 别把JMS想复杂但也要心怀敬畏我给初学者的建议是先用Boot自动配置把一条消息从生产者发到消费者跑通一条完整链路然后尝试发一个自定义对象、接一个监听方法最后再把连接池、监听器工厂、幂等这些生产要素逐步加进去。这套渐进式的路径比我当年一上来就啃规范、研究各种传输协议要高效得多。6.2 我的几个切身体会第一点消息系统的设计要从“最开始就要考虑失败”。很多系统在开发阶段完全没考虑过“消费者宕机了怎么办”“消息消费失败要不要重试”等到线上出事才手忙脚乱。建议从第一天起就做好幂等和死信队列这两笔投入的性价比极高。第二点不要轻易关闭持久化。有人说消息丢了就丢了反正业务上可以补偿。但在金融、订单这类场景丢消息不是小问题。哪怕为了性能也要先把持久化开着然后靠批量消费和并发调优来提升吞吐。为了性能牺牲可靠性是我见过的最危险的一种取舍。第三点监控一定要提前配。消息积压不像接口超时那样能立即被用户感知到它是慢慢恶化、等到明显的时候往往已经积压了几个小时。提前做好积压告警能帮你把很多“事故”扼杀在“故障”阶段。我在实际维护消息系统的过程中越来越觉得——让消息系统跑起来并不难难的是跑起来之后还能在你睡觉的时候自觉把活干好、把异常报告出来。而这恰恰是Spring集成JMS最有价值的地方框架把繁琐的底层细节承担了我们把业务可靠性和监控兜底做好整个系统就能稳定地异步运转起来了。