Spring集成JMS实战：消息中间件、异步解耦与事务可靠性

1. 理解JMS集成里的四个角色1.1 为什么需要JMS从同步调用到异步解耦先聊一个很多同学在项目里都会遇到的场景用户注册成功后需要发送欢迎邮件、赠送积分、更新统计报表。如果这些操作全部注册完同步执行一次注册请求可能要拖上2到3秒用户体验很差而且邮件服务一旦出问题注册接口直接就跟着挂掉了。解决这个问题最经典的做法就是把注册成功的消息丢到消息队列里由专门的消费者慢慢处理。Java消息服务JMS就是Java领域处理这类异步消息的标准规范它定义了消息的创建、发送、接收和读取的一套统一API至于底层用什么实现——是ActiveMQ、RabbitMQ的AMQP协议还是Tibco、IBM MQ对上层业务代码可以不透明。Spring在这一层做了非常关键的封装它把JMS繁杂的Connection、Session、MessageProducer这些原生API收敛成了模板类和注解。你在项目里只要能注入一个JmsTemplate一行代码就能把消息发出去接收端只要写一个标注了JmsListener的方法消息到了自动触发业务逻辑。对于初学者来说刚接触JMS原生API时经常会被Connection如何复用、Session要不要关闭、怎么处理消息确认这些细节堵住Spring把这些东西全部接管之后学习曲线其实被拉得非常平缓。这套集成意味着业务系统内部模块之间的通信不再依赖同步HTTP调用而是通过消息中间件解耦。生产者只管把消息发到目的地消费者按自己的节奏消费。这样做的收益是立竿见影的高并发时消息堆积在中间件里不会被直接压垮某个消费服务重启或者升级消息也不会丢失恢复后自动继续处理。1.2 JMS中必须搞懂的三组核心概念聊Spring集成之前我建议先把原生JMS里的几个底层概念扫一眼否则Spring封装虽然用起来简单出问题排查时还是会抓瞎。第一组是连接相关概念ConnectionFactory用来创建物理连接Connection代表到中间件的会话通道Session在JMS里是进行消息收发的工作单元类似JDBC的Connection级别的概念。Spring提供连接池管理这些资源避免每个操作都新建连接否则性能会非常难看。第二组是消息模型相关概念消息目的地分两种Queue队列和Topic主题。队列是点对点模型一条消息只被一个消费者消费消费完之后消息就从队列中移除Topic是发布订阅模型一个消息可以广播给所有订阅者。发送方不需要关心对方是否在线接收方也不需要知道是谁发的这就是松耦合的核心。第三组是消息结构JMS消息由消息头、消息属性和消息体组成。消息体有TextMessage、MapMessage、ObjectMessage、BytesMessage、StreamMessage五种类型日常用得最多的是TextMessage和MapMessage。Spring集成JMS时做的第一个重要决定就是你不需要自己手动创建Session、生产者这些原生对象而是把ConnectionFactory交给Spring管理然后基于它创建JmsTemplate。JmsTemplate每次调用send或者convertAndSend内部自动完成获取连接、创建会话、发送消息、提交事务、释放资源这一整套生命周期。这个设计思路和JdbcTemplate一脉相承都是模板方法模式在Spring生态里的经典应用。2. 环境准备从零搭建一个可运行的集成环境2.1 依赖引入与版本选择现在主流项目基本都是基于Spring Boot来开发的所以集成JMS最舒服的方式就是引入官方起步依赖。我以ActiveMQ作为消息中间件举例因为它是纯Java实现下载解压就能启动最适合学习和本地开发。如果是公司生产环境用RabbitMQ对方也支持JMS规范部分配置差异我会在后面的章节里点出来。如果用的是Maven第一个要加的是Spring Boot的JMS起步依赖它会把spring-jms、spring-messaging这些核心库以及ActiveMQ客户端库自动带进来。在Spring Boot 2.x和3.x中坐标写法有一点区别但核心引入方式一致dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-activemq/artifactId /dependency加上这个依赖之后类路径里就会自动配置一个ActiveMQ连接工厂。Spring Boot的自动化配置逻辑是只要检测到类路径里存在ActiveMQ相关的类就会默认创建一个连接工厂Bean。如果你用的是旧版Spring Boot 1.x对应的坐标名是spring-boot-starter-activemq但默认配置的broker URL可能不同这块要留意版本差异。这里我想强调一点Spring Boot的自动配置虽然省事但不会帮你注册队列和主题的Bean。原因是它没办法知道你业务里要往哪个目的地发消息这部分需要开发者自己声明。2.2 配置连接工厂和消息目的地在application.yml里最基础的配置项就是把Broker地址指到本地ActiveMQ的默认端口61616以及设置用户名和密码。ActiveMQ默认的管理员账号密码是admin/admin生产环境中这个配置必须改成强密码而且要限制对外开放否则就是网络层面的裸奔。spring: activemq: broker-url: tcp://127.0.0.1:61616 user: admin password: admin pool: enabled: true max-connections: 10Spring Boot 2.x之后官方建议开启连接池底层用的是连接池实现来重用连接、会话和生产者对象。暴露到配置项上就是pool.enabledtrue。不做这一步的话每次发送消息都会创建新连接性能会呈数量级下降。接着声明目的地。可以用注解方式声明一个Queue或Topic这种方式在配置集中管理场景下非常清晰Configuration public class JmsConfig { Bean public Queue orderQueue() { return new ActiveMQQueue(order.queue); } Bean public Topic orderTopic() { return new ActiveMQTopic(order.topic); } }这种方案的优点是声明式、依赖注入比较稳业务代码里直接注入Queue就能用。另一种更灵活的做法是运行时动态生成目的地也就是把目的地名称作为参数传入ActiveMQ内部根据名称自动创建Queue或Topic。这种方式适合消息目的地方案不固定的业务但在集群部署时可能会造成大量自动创建的队列堆积需要运维层面做清理策略。这里有几个配置细节需要提醒。第一个是队列名和主题名的规范建议统一前缀比如业务名加功能名再加queue/topic后缀千万不要随手写个abc就完事否则生产环境上百个队列根本没法管理。第二个是ActiveMQ的auto-create默认打开哪怕你点了不存在的队列也会自动建出来开发方便生产环境建议关闭。我个人还会建议把pool配置里max-sessions-per-connection也关注一下尤其在预计并发量较大的场景合理的会话数是保障吞吐量的关键。2.3 JmsTemplate的实例化与默认行为Spring Boot自动配置会提供一个JmsTemplate实例直接注入使用即可。但这里有一个容易踩的坑默认JmsTemplate通过ActiveMQ连接工厂创建会话时如果连接工厂的clientId没设置而在消费端又启用了持久订阅就会报错。所以我在配置类里经常自己重新定义JmsTemplate而不是完全依赖默认配置Bean public JmsTemplate jmsTemplate(ConnectionFactory connectionFactory) { JmsTemplate jmsTemplate new JmsTemplate(connectionFactory); jmsTemplate.setDeliveryMode(DeliveryMode.PERSISTENT); jmsTemplate.setDeliveryDelay(0L); return jmsTemplate; }实际上Spring Boot自动配置的JmsTemplate已经足够覆盖90%的基础场景它默认的消息投递模式是持久化而且默认开启了Qos设置也就是根据JmsTemplate自身的设置覆盖消息属性。这里特别要留意的属性是setExplicitQosEnabled这个属性控制是否使用JmsTemplate上配置的投递模式、优先级、存活时间覆盖消息自带的属性。默认情况下如果你自己创建了一个消息对象并在里面设置了-delivery模式发送时可能被JmsTemplate覆盖这个行为要通过显式开启或关闭来控制。我见过不少同学在这里调了很久觉得消息发出去了应该马上就能被消费结果发现消息在队列里躺着不动就是因为投递模式或者优先级设置被覆盖导致行为不一致。这种情况检查的方向就是JmsTemplate的Qos配置和消息属性这两层。3. 核心代码实现消息发送与消费3.1 使用JmsTemplate发送消息的三种姿势先看最简单的调用方式直接发送文本内容到队列。JmsTemplate封装得非常彻底连消息对象都不用我们手动创建Service public class OrderMessageSender { private final JmsTemplate jmsTemplate; private final Queue orderQueue; public OrderMessageSender(JmsTemplate jmsTemplate, Queue orderQueue) { this.jmsTemplate jmsTemplate; this.orderQueue orderQueue; } public void sendOrder(String orderId) { jmsTemplate.convertAndSend(orderQueue, orderId); } }convertAndSend会把传入的普通Java对象自动转换成TextMessage省掉了手工创建消息的步骤。如果业务上需要在消息里附带一些额外属性比如消息的优先级、业务标识、幂等键可以这样操作public void sendOrderWithProperties(String orderId) { jmsTemplate.convertAndSend(orderQueue, orderId, message - { message.setStringProperty(businessType, order); message.setStringProperty(orderId, orderId); message.setIntProperty(retryCount, 0); return message; }); }第二个参数是Map或对象时convertAndSend会走MessageConverter转换机制。默认的SimpleMessageConverter只支持String、Map、byte[]、Serializable对象这几类。想传自定义POJO最简单的办法是让POJO实现Serializable接口Spring会帮你序列化成ObjectMessage。这种方式实现成本最低但跨语言解析会有点麻烦如果多个系统用不同语言更推荐转成JSON字符串发送那边拿到后再解析。第三种姿势是发送到Topic实现广播。Topic的场景是一个消息被多个订阅者同时收到所以发送方通常不关注谁在消费只需要把它send到Topic目的地所有有效订阅者都能收到消息。public void publishOrderEvent(String orderId) { jmsTemplate.convertAndSend(orderTopic, 订单创建成功: orderId); }这里提醒一下JmsTemplate的send和convertAndSend重载非常多如果只传消息内容不传目的地会使用默认目的地。默认目的地需要在JmsTemplate上setDefaultDestination否则调用时直接抛异常不熟悉的人第一次看到IllegalStateException时会摸不着头脑。3.2 JmsListener注解式消费消息消费端最推荐的做法就是注解驱动定义一个监听器方法Spring在消息到达时自动调用这个方法。Component public class OrderMessageConsumer { JmsListener(destination order.queue) public void onOrderMessage(String orderId) { System.out.println(收到订单消息: orderId); // 处理业务逻辑 } }看到这里有些人会好奇为什么方法参数直接就能收到String因为Spring内置了一个消息转换器它会把JMS的TextMessage自动转换成String类型。如果是MapMessage参数可以声明为Map类型。需要拿消息头或属性的时候方法参数后面加一个Message类型参数就行JmsListener(destination order.queue) public void onOrderMessage(String orderId, Message message) throws JMSException { String businessType message.getStringProperty(businessType); String msgId message.getJMSMessageID(); // 幂等处理时JMSMessageID非常适合用来做数据库唯一键 }JmsListener默认情况下线程调度由默认的消息监听容器处理这意味着如果你同时注册了多个监听器方法它们的并发配置都绑定在同一个容器工厂上。更细粒度的控制可以通过自定义容器工厂来实现后面讲监听器容器时再展开。同时要注意JmsListener所在的方法异常时Spring会不断重试还是直接放弃取决于你配置的消息确认模式和监听器容器异常处理器。默认的DefaultMessageListenerContainer会不断把消息重新投递给监听器也就是说如果你的业务代码抛出异常消息不会被确认中间件会再次投递这条消息直到成功或者被专门处理。这个机制既要利用好重试保障不掉消息又要防止无限重试导致死循环。后面我会讲怎么配合重试策略和死信队列来收尾。除了注解方式传统的同步接收方式也仍然可用。通过JmsTemplate的receive方法手动从队列里拿一条消息处理。这种模式适合批处理任务或者需要自己控制消费速度的场景但方式比较古老。3.3 receive模式与事务性代码示例某些业务场景下比如报表批量计算任务希望主动去队列里拉消息而不是被动监听这时可以这样public OrderDTO receiveOrder() { Message message jmsTemplate.receive(orderQueue, 3000); if (message instanceof TextMessage) { return parseOrder(((TextMessage) message).getText()); } return null; }receive的重载方法里第二个参数是等待消息的超时时间单位是毫秒。这里有个很重要的坑如果在同一个线程里先调用了receive又启动了其他事务性操作会直接影响事务边界的划定。最好是让receive和对应的业务处理处在同一个事务会话里可以用sessionTransacted设为true来开启一个本地事务在这个事务里接收消息并执行数据库操作如果后续数据库操作失败消息不会确认可直接投递给其他消费者这就实现了消息和数据库操作的原子性。关于事务部分Spring也提供了一套注解驱动方案。在监听方法上直接加Transactional配合连接工厂和事务管理器就可以保证消息接收与业务数据库操作在同一个事务里提交或回滚。这个过程我在第四章节详细分析。4. 事务、可靠性与监听器容器原理4.1 消息确认机制从AUTO_ACKNOWLEDGE到CLIENT_ACKNOWLEDGEJMS定义了多种消息确认模式理解它们是学习可靠性的关键。最常用的是Session.AUTO_ACKNOWLEDGE消息在消费方法正常返回后自动确认如果抛异常则不确认消息会重新投递。这是默认监听器容器的默认行为对大多数业务足够了。另一个是Session.CLIENT_ACKNOWLEDGE消息只有客户端显式调用message.acknowledge()之后才确认。这种模式适合业务逻辑比较长的场景比如你拉取消息后要花很久做后续处理希望确认时机自己掌握。还有一种模式是DUPS_OK_ACKNOWLEDGE它允许重复确认消息可能重复投递换来的是更高的吞吐量。在实际项目中用它之前必须保证消费逻辑是幂等的否则数据和缓存都会出现不一致。Spring Boot中配置确认模式比较直接在application.yml里设置spring.jms.listener.acknowledge-mode即可。但要特别注意的是不同的监听器容器对这个配置的响应并不完全一致。比如使用DefaultMessageListenerContainer时即使你配置了CLIENT_ACKNOWLEDGE容器内部还是会用会话级确认机制因为容器对会话做了托管。这个概念对新手来说很绕我在实际项目里给出的建议是默认用AUTO_ACKNOWLEDGE除非你很清楚自己在做什么。需要补充的是Transacted会话。当你把Session设置为事务性会话之后确认模式基本上就退居二线了因为事务提交本身就意味着确认。这就是为什么Spring文档总是说本地事务模式下的可靠性最高消息处理与数据库更新可以保持一致性。4.2 sessionTransacted与外部事务管理器如果只是消费消息后做一些内存操作把监听器容器上的sessionTransacted设为true就够了Session会被标记为事务性Spring会在监听器方法返回后提交这个事务如果方法抛出异常就回滚并重新投递消息。但如果消费过程中要操作数据库这个事务就没有覆盖到数据库操作了。这时要用外部事务管理器配置一个JmsTransactionManager并和Spring的Transactional结合使用。JmsTransactionManager做的事情是把JMS会话和Spring事务绑定起来事务提交时数据库操作和消息确认一起提交它本质上是一种本地事务比分布式事务要轻适用范围是单数据源单消息中间件的场景。配置方式是这样的Bean public JmsTransactionManager jmsTransactionManager(ConnectionFactory connectionFactory) { return new JmsTransactionManager(connectionFactory); }然后在监听方法上JmsListener(destination order.queue) Transactional public void onOrderMessage(OrderDTO order) { // 保存订单到数据库 orderRepository.save(order); // 更新积分等操作 integralService.addIntegral(order.getUserId(), 10); }这段代码最关键的意义是如果orderRepository.save成功了但后续积分操作抛异常整个事务回滚消息不会被确认会重新进入队列等待下一次消费。这样数据库和消息消费之间不会出现各走各的路的情况。不过要提醒一点一旦启用事务消息消费的吞吐量会受到影响因为每个事务都需要提交而且回滚时会重新投递消息。业务上如果允许消息少量重复处理就先把幂等做好再追求吞吐量。4.3 DefaultMessageListenerContainer的行为与并发配置Spring集成JMS的核心监听容器有两个实现一个是SimpleMessageListenerContainer另一个是DefaultMessageListenerContainer。后者非常强大它自己管理消费者线程、动态调整并发数、参与Spring事务、支持暂停恢复绝大多数场景都用它。默认情况下一个JmsListener对应一个消费者线程这对并发量高的业务来说是不够的。可以自定义一个容器工厂来设置并发参数Bean public JmsListenerContainerFactory? queueListenerFactory( ConnectionFactory connectionFactory, DefaultJmsListenerContainerFactoryConfigurer configurer) { DefaultJmsListenerContainerFactory factory new DefaultJmsListenerContainerFactory(); configurer.configure(factory, connectionFactory); factory.setConcurrency(3-10); return factory; }concurrency设置的格式是最小-最大并发消费者数量比如3-10代表最少3个消费线程最多根据消息负载动态扩展到10个。对应到监听注解上JmsListener(destination order.queue, containerFactory queueListenerFactory) public void onOrderMessage(String orderId) { // ... }并发消费者数量并不是越大越好要考虑到下游数据库连接池大小、中间件性能以及消息处理耗时。处理耗时为50毫秒时单线程每秒20条5个线程就能到100条每秒如果处理耗时为1秒20个线程的实际效果更明显。调整并发参数时重点看消费者线程是否一直处于忙碌状态以及消息积压是否在下降。还要注意一个容易被忽略的点就是并发消费者和消息顺序性的关系。JMS规范本身不保证全局消息顺序但同一个生产者发送的消息在只有一个消费者的情况下通常能保持顺序。一旦并发消费者数量大于1消息被多个线程并行消费顺序就完全无法保证了。如果你的业务强依赖消息顺序并发数只能设1或者从消息里带一个序号字段由消费端自行排序。5. 常见问题排查与经验锦囊5.1 消息消费不到人监听器不触发的排查链路这是我在群里被问过最多的问题JmsListener方法写好了也启动项目了但发消息之后监听方法一直没反应。排查的第一件事就是类路径问题确认spring-boot-starter-activemq的依赖真的加进来了并且没有同时存在多个JMS实现类导致自动配置混乱。特别是有些项目里同时引入了老版本activemq-all和新版客户端jar会产生类冲突启动时就能看到诡异异常。第二件事是确认EnableJms注解是否存在。Spring Boot在检测到JMS起步依赖时通常会自动开启但在某些自定义配置场景下你如果覆盖了JmsListenerContainerFactory可能会破坏掉这个自动开启行为。稳妥做法是自己在配置类上显式加EnableJms。第三件事是确认监听器容器工厂被正确关联。如果项目里有多个容器工厂JmsListener的containerFactory参数没写Spring会去找默认的工厂而默认工厂可能因为连接配置差异连不上中间件这会导致监听容器不断重连日志里会有警告但是不熟悉的人看不到。第四件事也常见application.yml里的broker-url配置正确指向了中间件地址但中间件所在的机器防火墙没开放61616端口客户端连接被拒绝。服务已经启动了日志也没疯狂打印错误只是每隔几秒重试一次看起来就像监听不触发。看到这里先试试用命令telnet测试一下端口比盯着日志更直接。5.2 消息重复消费与幂等设计JMS本身就属于At Least Once语义的消息系统消息重复是常态不是异常。中间件在消费者端超时未确认时会重新投递消息网络抖动、消费者进程被杀都可能导致重复消费。所以消费端必须做幂等。常见的幂等方案是基于业务唯一键加数据库唯一索引。比如订单消息里带orderId收到消息后先查一下本地表里有没有这张订单有就直接跳过没有就插入如果插入时遇到唯一键冲突就说明是重复消息catch住异常往下走。这种方式在分布式环境里也是成立的因为唯一索引的约束力不受多个实例的影响。如果想用Message自身的ID做去重需要注意JMSMessageID在不同中间件中可能不保证全局唯一而且同一个消息被重投时JMSMessageID大概率保持不变但跨会话或跨broker场景要谨慎。更保险的做法是把业务状态字段提取出来用你自己生成的业务事件ID做幂等键这个逻辑跟用的是MQ还是JMS没有关系。5.3 事务不回滚的三种成因JmsListener加Transactional后业务方法抛异常但消息还是被确认了这个现象通常有三种成因。第一种是注解被类内部自身调用绕过。Spring事务是通过代理实现的如果监听方法里通过this.xxx()调用另一个自己的方法事务注解不会生效因为调用发生在原生对象上而不是代理对象上。这个坑在几乎所有Spring事务教程里都出现过但换到JMS场景中仍然有人中招。第二种是事务管理器和监听容器没有正确关联。JmsTransactionManager确实定义了但容器工厂里没有显式setTransactionManagerSpring Boot自动配置在某些情况下会使用默认决策结果实际执行时数据库操作和消息确认各走各的事务。检查的方式就是看配置类里容器工厂是否手动设置了transactionManager。第三种是异常类型是Error而不是Exception或者异常已经被方法内部捕获。Spring事务默认只有RuntimeException和Error触发回滚如果业务方法catch住了异常并且打印一行日志后正常返回那消息肯定会被确认整个业务看起来是成功了。这一点在开发初期特别容易出问题我觉得比较靠谱的做法是让事务方法不要吞异常或者使用Transactional(rollbackFor Exception.class)明确指定回滚范围。5.4 消息堆积与消费吞吐量优化消息一直积压队列里数量越来越多除了消费者处理速度本身就慢之外还有几个常见原因。消费者并发数设置太小是其中一个刚才提到concurrency属性的动态调整可以让容器根据负载增加消费线程。但要注意的是并发数调大之后如果数据库连接池的数量不够会暴露连接池瓶颈到时候消息消费变慢数据库连接等待时间陡增需要一起评估。另一个原因是每消费一条消息执行了耗时的外部调用而且这个外部调用没有设置超时时间。外部服务一旦变慢整个消费线程就全部卡在等待上队列积压越来越严重。针对这种情况我一般在消费逻辑里把外部调用统一加上超时控制并配合线程池隔离防止一个下游服务拖垮整个消息处理链路。再有一个容易被忽略的点是消息大小。如果一条消息体特别大比如几MB的JSON序列化和网络传输的开销会明显上升。ActiveMQ默认有消息大小限制超过限制的消息要么拒绝要么持久化失败。如果业务上必须传输较大的数据建议消息里只放一个业务标识或者文件路径真正的数据传输走文件或数据库这样消息系统会更加轻快、稳定。6. 从单机到集群生产环境落地的几个注意点说到生产环境很多人本地调试得好好的一到线上就出问题原因往往不是代码而是使用方式不对。集成JMS在集群环境下有几个值得提前规划的点。第一个是Broker集群的选择和网络拓扑。ActiveMQ支持网络连接器把多个Broker连成一个网络也支持共享存储的主从模式。不同模式下的连接失败重连行为差异很大消费端使用failover协议时要注意failover传输层会把多个Broker地址串起来自动切换spring: activemq: broker-url: failover:(tcp://node1:61616,tcp://node2:61616)?randomizefalsefailover协议对客户端来说透明但有一点必须提醒如果集群中的消息还没同步完成消费者正好切到另一个节点可能出现短暂的消息不可见或重复投递。配合消费端的幂等设计这个问题可以被消化。第二个是持久化消息的存储策略。ActiveMQ默认的KahaDB适合大多数场景但高频高吞吐时日志文件可能增长很快建议对存储目录单独设置磁盘空间告警。同时消息长时间积压不消费也会撑爆存储运维上需要有清理机制。第三个是监控。生产环境没有监控等于盲飞要让消息中间件的队列深度、连接数、消息吞吐量指标都接入监控体系。队列积压有阈值告警消费者线程长时间不动需要被看见。我在项目里还会把JmsListener的消费耗时和消费失败次数记录为业务指标这样消息模块好不好看数据可视化面板就够了。从代码层面来说无论本地还是生产发送消息后一定养成打日志的习惯尤其是关键消息的消息ID和业务ID。排查和追踪问题时没有消息ID就去中间件后台一个个翻效率低得出奇。7. 写在最后的一点心得集成JMS本身不难核心就三件事目的地声明、发送模板、监听容器。真正决定成败的是工程化细节比如确认机制是否和你的事务边界匹配、并发数跟下游资源是否匹配、消费逻辑是否幂等。我个人在实际项目里的习惯是刚上手时先用最简单的AUTO_ACKNOWLEDGE跑通全链路然后在关键业务上再逐步引入事务、并发调优和监控告警千万不要一上来就堆配置因为配置越复杂出问题时需要排查的维度就越多。如果你现在正准备做项目里的异步化改造不妨从下单通知、积分变动这类业务开始试点把Spring集成JMS这套基础链路跑通再慢慢往可靠性要求更高的业务上推广。消息中间件在架构里的定位是骨架它的稳定性影响全局所以每一条消息策略都应该有明确的业务理由而不是盲目跟风。