Java ActiveMQ 实战指南：从 JMS 队列、主题到 Spring Boot 消息处理

简介ActiveMQ是 Apache 提供的消息中间件。它在 Java 生态里很常见尤其是传统企业系统、老项目改造、JMS 规范相关系统中出现得比较多。简单理解生产者 | v ActiveMQ Broker | v 消费者ActiveMQ 最常见的使用方式是 JMS。JMS 全称是Java Message Service它是一套 Java 消息服务 API 规范。ActiveMQ 和 JMS 的关系可以类比成JMS 是规范 ActiveMQ 是实现一句话概括ActiveMQ 是 Java 生态里经典的 JMS 消息中间件适合异步处理、系统解耦、任务队列和发布订阅场景。ActiveMQ 解决什么问题一个业务动作经常会带出很多后续动作。比如订单创建成功后扣库存 发短信 发邮件 写日志 生成物流任务 同步积分如果全部同步执行接口会变慢。如果短信服务异常订单接口也可能被影响。使用 ActiveMQ 后可以拆成订单服务 | v 发送订单消息 | v ActiveMQ | -- 库存服务消费消息 -- 短信服务消费消息 -- 邮件服务消费消息 -- 日志服务消费消息常见作用异步处理系统解耦削峰填谷任务分发发布订阅消息重试和失败兜底ActiveMQ Classic 和 ArtemisActiveMQ 需要先区分两条线名称说明ActiveMQ Classic传统 ActiveMQ 5.x / 6.x 线老项目和 JMS 系统常见ActiveMQ Artemis新一代消息 Broker性能和架构更现代简单理解ActiveMQ Classic经典 JMS Broker ActiveMQ Artemis新一代多协议、高性能 Broker如果是维护老系统经常会遇到 ActiveMQ Classic。如果是新系统选型并且明确要用 Apache ActiveMQ 体系Artemis 更值得优先评估。本文示例以 ActiveMQ Classic Spring Boot JMS 为主因为它更贴近日常 Java 老系统和企业项目集成场景。ActiveMQ 和 RabbitMQ 的区别对比项ActiveMQRabbitMQ常见协议JMS、OpenWire、STOMP、AMQP、MQTTAMQP 为主也支持多协议Java 规范关系JMS 体系很强不以 JMS 为核心核心模型Queue / TopicExchange / Queue / Binding路由能力JMS 模型更直接Exchange 路由更灵活常见场景传统企业系统、JMS 项目微服务消息、灵活路由、事件分发ActiveMQ 更像Queue / Topic 模型清晰适合 JMS 项目。RabbitMQ 更像Exchange Routing Key 更灵活适合复杂路由。JMS 核心概念概念说明ConnectionFactory连接工厂Connection客户端到 Broker 的连接Session会话负责生产、消费、确认、事务Destination目的地Queue 或 TopicQueue点对点队列Topic发布订阅主题MessageProducer消息生产者MessageConsumer消息消费者Message消息对象整体流程ConnectionFactory | v Connection | v Session | -- MessageProducer - Destination | -- MessageConsumer - DestinationQueue 点对点模式Queue 是点对点模式。特点一条消息只会被一个消费者处理。示例order.queue | -- consumer-1 -- consumer-2如果order.queue有 10 条消息两个消费者会竞争消费。每条消息最终只会被其中一个消费者处理。适合订单处理短信发送邮件发送报表生成后台任务Topic 发布订阅模式Topic 是发布订阅模式。特点一条消息可以被多个订阅者收到。示例order.topic | -- sms-consumer -- email-consumer -- log-consumer订单支付成功消息发送到order.topic后短信、邮件、日志服务都能收到。适合业务事件广播通知消息日志订阅缓存刷新Topic 持久订阅和非持久订阅Topic 需要注意订阅方式。类型离线期间消息是否保留说明非持久订阅不保留消费者在线时才能收到持久订阅保留消费者离线后再上线也能收到非持久订阅适合临时通知。持久订阅适合关键业务事件。持久订阅通常需要设置唯一的clientId和订阅名。Docker 启动 ActiveMQ Classic本地学习可以使用 Docker。不同镜像维护方式不同实际项目以团队镜像仓库或官方发布包为准。示例docker run -d \ --name activemq-classic \ -p 61616:61616 \ -p 8161:8161 \ apache/activemq-classic:latest端口说明端口作用61616JMS / OpenWire 连接端口8161Web 管理控制台管理后台http://localhost:8161/admin常见默认账号admin / admin如果镜像没有提供默认账号需要查看镜像说明或自定义配置。原生 JMS 依赖Spring Boot 3 使用jakarta.jms包名。如果使用 ActiveMQ Classic 6 或 Jakarta 体系可以选择 Jakarta 版本客户端。dependency groupIdorg.apache.activemq/groupId artifactIdactivemq-client-jakarta/artifactId version${activemq.version}/version /dependency如果是老项目可能仍然使用javax.jms和 ActiveMQ 5.x 客户端。dependency groupIdorg.apache.activemq/groupId artifactIdactivemq-client/artifactId version${activemq.version}/version /dependency包名差异老项目javax.jms.* 新项目jakarta.jms.*原生 JMS发送 Queue 消息package com.example.activemq.raw; import jakarta.jms.Connection; import jakarta.jms.ConnectionFactory; import jakarta.jms.DeliveryMode; import jakarta.jms.Destination; import jakarta.jms.MessageProducer; import jakarta.jms.Session; import jakarta.jms.TextMessage; import org.apache.activemq.ActiveMQConnectionFactory; public class RawQueueProducer { private static final String BROKER_URL tcp://localhost:61616; private static final String QUEUE_NAME order.queue; public static void main(String[] args) throws Exception { ConnectionFactory factory new ActiveMQConnectionFactory(BROKER_URL); try (Connection connection factory.createConnection(); Session session connection.createSession(false, Session.AUTO_ACKNOWLEDGE)) { Destination destination session.createQueue(QUEUE_NAME); MessageProducer producer session.createProducer(destination); producer.setDeliveryMode(DeliveryMode.PERSISTENT); TextMessage message session.createTextMessage(订单创建成功); message.setStringProperty(bizType, order); message.setStringProperty(messageId, msg-1001); producer.send(message); System.out.println(消息已发送 message.getText()); } } }说明Session.AUTO_ACKNOWLEDGE自动确认 DeliveryMode.PERSISTENT持久化消息 createQueue创建 Queue 目的地原生 JMS消费 Queue 消息package com.example.activemq.raw; import jakarta.jms.Connection; import jakarta.jms.ConnectionFactory; import jakarta.jms.MessageConsumer; import jakarta.jms.Session; import jakarta.jms.TextMessage; import org.apache.activemq.ActiveMQConnectionFactory; public class RawQueueConsumer { private static final String BROKER_URL tcp://localhost:61616; private static final String QUEUE_NAME order.queue; public static void main(String[] args) throws Exception { ConnectionFactory factory new ActiveMQConnectionFactory(BROKER_URL); Connection connection factory.createConnection(); Session session connection.createSession(false, Session.CLIENT_ACKNOWLEDGE); MessageConsumer consumer session.createConsumer(session.createQueue(QUEUE_NAME)); connection.start(); consumer.setMessageListener(message - { try { if (message instanceof TextMessage textMessage) { System.out.println(收到消息 textMessage.getText()); } message.acknowledge(); } catch (Exception e) { throw new RuntimeException(e); } }); System.out.println(消费者已启动); Thread.currentThread().join(); } }CLIENT_ACKNOWLEDGE表示客户端手动确认。注意JMS 的 acknowledge() 可能确认当前 Session 中已消费的一批消息。如果需要更强的边界控制可以考虑事务会话。JMS 确认模式JMS 常见确认模式模式说明AUTO_ACKNOWLEDGE监听器正常返回后自动确认CLIENT_ACKNOWLEDGE客户端调用acknowledge()确认DUPS_OK_ACKNOWLEDGE允许延迟确认可能出现重复SESSION_TRANSACTED使用事务提交或回滚AUTO_ACKNOWLEDGE使用简单。关键业务更常见的是事务或手动确认。JMS 事务创建事务会话Session session connection.createSession(true, Session.SESSION_TRANSACTED);处理成功session.commit();处理失败session.rollback();示例MessageConsumer consumer session.createConsumer(session.createQueue(order.queue)); TextMessage message (TextMessage) consumer.receive(); try { System.out.println(处理消息 message.getText()); // 执行业务逻辑 session.commit(); } catch (Exception e) { session.rollback(); }事务回滚后消息会重新投递。Spring Boot 依赖Spring Boot 集成 ActiveMQ Classicdependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-activemq/artifactId /dependency如果选择 Artemisdependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-artemis/artifactId /dependencyClassic 和 Artemis 不建议在同一个应用里混用。application.ymlActiveMQ Classic 示例spring: activemq: broker-url: tcp://localhost:61616 user: admin password: admin jms: listener: acknowledge-mode: auto如果需要连接池可以配置连接池依赖和对应参数。不同 Spring Boot 版本对连接池自动配置细节略有差异实际项目以当前版本文档为准。开启 JMS 注解使用JmsListener需要开启 JMS 注解支持。package com.example.activemq.config; import org.springframework.context.annotation.Configuration; import org.springframework.jms.annotation.EnableJms; Configuration EnableJms public class JmsConfig { }消息对象package com.example.activemq.order; import java.math.BigDecimal; import java.time.LocalDateTime; public record OrderCreatedMessage( Long orderId, String orderNo, Long userId, BigDecimal amount, LocalDateTime createdAt ) { }JSON 消息转换器业务系统里更常用 JSON而不是 Java 原生对象序列化。package com.example.activemq.config; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.jms.support.converter.MappingJackson2MessageConverter; import org.springframework.jms.support.converter.MessageConverter; import org.springframework.jms.support.converter.MessageType; Configuration public class JmsMessageConfig { Bean public MessageConverter jacksonJmsMessageConverter() { MappingJackson2MessageConverter converter new MappingJackson2MessageConverter(); converter.setTargetType(MessageType.TEXT); converter.setTypeIdPropertyName(_type); return converter; } }Queue 实战订单消息常量package com.example.activemq.config; public class ActiveMqDestinations { public static final String ORDER_QUEUE order.queue; public static final String NOTICE_TOPIC notice.topic; private ActiveMqDestinations() { } }生产者package com.example.activemq.order; import com.example.activemq.config.ActiveMqDestinations; import org.springframework.jms.core.JmsTemplate; import org.springframework.stereotype.Service; Service public class OrderMessageProducer { private final JmsTemplate jmsTemplate; public OrderMessageProducer(JmsTemplate jmsTemplate) { this.jmsTemplate jmsTemplate; } public void sendOrderCreated(OrderCreatedMessage message) { jmsTemplate.convertAndSend(ActiveMqDestinations.ORDER_QUEUE, message); } }消费者package com.example.activemq.order; import com.example.activemq.config.ActiveMqDestinations; import org.springframework.jms.annotation.JmsListener; import org.springframework.stereotype.Component; Component public class OrderMessageConsumer { JmsListener(destination ActiveMqDestinations.ORDER_QUEUE) public void handle(OrderCreatedMessage message) { System.out.println(收到订单消息 message); // 执行业务逻辑扣库存、发通知、写日志等 } }Controllerpackage com.example.activemq.order; import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; import java.math.BigDecimal; import java.time.LocalDateTime; RestController RequestMapping(/api/orders) public class OrderController { private final OrderMessageProducer producer; public OrderController(OrderMessageProducer producer) { this.producer producer; } PostMapping public String createOrder() { OrderCreatedMessage message new OrderCreatedMessage( 1001L, O202606300001, 2001L, new BigDecimal(99.80), LocalDateTime.now() ); producer.sendOrderCreated(message); return 订单消息已发送; } }访问POST http://localhost:8080/api/orders同时支持 Queue 和 TopicJmsTemplate有一个关键属性pubSubDomain含义falseQueue 模式 trueTopic 模式如果一个项目同时使用 Queue 和 Topic可以分别创建两个模板。package com.example.activemq.config; import jakarta.jms.ConnectionFactory; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.jms.config.DefaultJmsListenerContainerFactory; import org.springframework.jms.core.JmsTemplate; Configuration public class JmsTemplateConfig { Bean public JmsTemplate queueJmsTemplate(ConnectionFactory connectionFactory) { JmsTemplate template new JmsTemplate(connectionFactory); template.setPubSubDomain(false); return template; } Bean public JmsTemplate topicJmsTemplate(ConnectionFactory connectionFactory) { JmsTemplate template new JmsTemplate(connectionFactory); template.setPubSubDomain(true); return template; } Bean public DefaultJmsListenerContainerFactory topicListenerContainerFactory( ConnectionFactory connectionFactory ) { DefaultJmsListenerContainerFactory factory new DefaultJmsListenerContainerFactory(); factory.setConnectionFactory(connectionFactory); factory.setPubSubDomain(true); return factory; } }Topic 实战广播通知生产者package com.example.activemq.notice; import com.example.activemq.config.ActiveMqDestinations; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.jms.core.JmsTemplate; import org.springframework.stereotype.Service; Service public class NoticeProducer { private final JmsTemplate topicJmsTemplate; public NoticeProducer(Qualifier(topicJmsTemplate) JmsTemplate topicJmsTemplate) { this.topicJmsTemplate topicJmsTemplate; } public void publish(String content) { topicJmsTemplate.convertAndSend(ActiveMqDestinations.NOTICE_TOPIC, content); } }短信消费者package com.example.activemq.notice; import com.example.activemq.config.ActiveMqDestinations; import org.springframework.jms.annotation.JmsListener; import org.springframework.stereotype.Component; Component public class SmsNoticeConsumer { JmsListener( destination ActiveMqDestinations.NOTICE_TOPIC, containerFactory topicListenerContainerFactory ) public void handle(String message) { System.out.println(短信服务收到通知 message); } }邮件消费者package com.example.activemq.notice; import com.example.activemq.config.ActiveMqDestinations; import org.springframework.jms.annotation.JmsListener; import org.springframework.stereotype.Component; Component public class EmailNoticeConsumer { JmsListener( destination ActiveMqDestinations.NOTICE_TOPIC, containerFactory topicListenerContainerFactory ) public void handle(String message) { System.out.println(邮件服务收到通知 message); } }一条 Topic 消息会被两个消费者都收到。Topic 持久订阅 Demo非持久订阅只接收在线期间的 Topic 消息。持久订阅需要设置clientId和订阅名。Bean public DefaultJmsListenerContainerFactory durableTopicListenerContainerFactory( ConnectionFactory connectionFactory ) { DefaultJmsListenerContainerFactory factory new DefaultJmsListenerContainerFactory(); factory.setConnectionFactory(connectionFactory); factory.setPubSubDomain(true); factory.setSubscriptionDurable(true); factory.setClientId(notice-service-01); return factory; }监听JmsListener( destination ActiveMqDestinations.NOTICE_TOPIC, subscription notice-service-subscription, containerFactory durableTopicListenerContainerFactory ) public void handleDurableTopic(String message) { System.out.println(持久订阅收到通知 message); }持久订阅注意点同一个 broker 上 clientId 需要唯一 同一个 clientId subscription 代表同一个持久订阅消息选择器消息选择器可以按消息属性过滤。发送jmsTemplate.convertAndSend(order.queue, message, jmsMessage - { jmsMessage.setStringProperty(bizType, order); jmsMessage.setStringProperty(level, important); return jmsMessage; });消费JmsListener( destination order.queue, selector bizType order AND level important ) public void handleImportantOrder(OrderCreatedMessage message) { System.out.println(收到重要订单消息 message); }选择器适合按少量稳定属性过滤。如果路由规则很复杂通常更适合拆分队列或 Topic。延迟消息ActiveMQ Classic 支持调度消息。常见属性属性作用AMQ_SCHEDULED_DELAY延迟多久投递AMQ_SCHEDULED_PERIOD重复投递周期AMQ_SCHEDULED_REPEAT重复次数AMQ_SCHEDULED_CRONCron 表达式发送延迟消息import org.apache.activemq.ScheduledMessage; jmsTemplate.convertAndSend(order.timeout.queue, message, jmsMessage - { jmsMessage.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, 30 * 60 * 1000L); return jmsMessage; });表示 30 分钟后投递。Broker 需要开启 scheduler 支持。死信队列ActiveMQ Classic 默认有死信队列。常见默认名称ActiveMQ.DLQ消息可能进入死信队列的常见原因消费者重复失败消息过期消息被回滚多次达到最大重投递次数ActiveMQ 常见处理链路消费者处理失败 | v Session rollback 或监听器抛异常 | v Broker 重新投递 | v 超过最大重投递次数 | v 进入 DLQ死信队列适合保留异常消息后台补偿处理人工排查告警通知Redelivery 重投递ActiveMQ 客户端可以设置重投递策略。import org.apache.activemq.ActiveMQConnectionFactory; import org.apache.activemq.RedeliveryPolicy; ActiveMQConnectionFactory factory new ActiveMQConnectionFactory(tcp://localhost:61616); RedeliveryPolicy policy new RedeliveryPolicy(); policy.setMaximumRedeliveries(3); policy.setInitialRedeliveryDelay(1000); policy.setRedeliveryDelay(2000); policy.setUseExponentialBackOff(true); policy.setBackOffMultiplier(2.0); factory.setRedeliveryPolicy(policy);含义配置说明maximumRedeliveries最大重投递次数initialRedeliveryDelay首次重投递延迟redeliveryDelay重投递间隔useExponentialBackOff是否指数退避backOffMultiplier退避倍数重投递次数耗尽后消息通常会进入 DLQ。Spring 事务消费Spring JMS 可以使用事务监听容器。Bean public DefaultJmsListenerContainerFactory txQueueListenerContainerFactory( ConnectionFactory connectionFactory ) { DefaultJmsListenerContainerFactory factory new DefaultJmsListenerContainerFactory(); factory.setConnectionFactory(connectionFactory); factory.setSessionTransacted(true); return factory; }监听JmsListener( destination ActiveMqDestinations.ORDER_QUEUE, containerFactory txQueueListenerContainerFactory ) public void handleWithTx(OrderCreatedMessage message) { System.out.println(事务消费消息 message); // 方法正常结束提交 Session // 方法抛出异常回滚 Session消息等待重投递 }事务消费适合消息处理成功才确认失败后触发重投递和本地数据库事务配合幂等处理消息中间件通常更接近“至少一次”投递。因此消费者需要能处理重复消息。常见做法消息带唯一 messageId 消费前检查处理记录 未处理则执行业务 处理成功后保存记录 重复消息直接跳过示例JmsListener(destination ActiveMqDestinations.ORDER_QUEUE) public void handle(OrderCreatedMessage message) { String messageId message.orderNo(); if (messageLogRepository.exists(messageId)) { return; } processOrder(message); messageLogRepository.save(messageId); }可以用来做幂等的方式数据库唯一索引消息消费日志表RedisSETNX业务状态机ActiveMQ 数据存储在哪里ActiveMQ 数据存储在 Broker 节点上。主要包括队列和主题元数据持久化消息订阅关系事务和调度消息信息ActiveMQ Classic 常见持久化存储是KahaDBArtemis 常见持久化方式是JournalDocker 部署时需要挂载数据目录。Classic 常见目录/opt/apache-activemq/data不同镜像目录可能不同需要以镜像说明为准。示例docker run -d \ --name activemq-classic \ -p 61616:61616 \ -p 8161:8161 \ -v activemq-data:/opt/apache-activemq/data \ apache/activemq-classic:latest如果没有挂载数据卷容器删除后数据也会删除。消息重启后是否保留取决于消息是否持久化 目的地和 Broker 持久化配置 存储目录是否保留管理后台常看指标ActiveMQ 管理后台常见关注点指标说明Pending Messages等待消费的消息Consumers消费者数量Enqueued入队消息总数Dequeued出队消息总数Dispatch Queue已分派但未确认消息Expired过期消息DLQ死信消息排查思路Pending Messages 持续增加消费能力不足或消费者异常 Consumers 为 0消费者未启动或监听配置错误 DLQ 增加业务处理失败、重投递耗尽或消息过期 Enqueued 高于 Dequeued生产速度超过消费速度常见使用建议Queue 和 Topic 分清楚Queue 是任务分发。一条消息只给一个消费者Topic 是广播事件。一条消息给多个订阅者如果业务是“发短信任务”通常选 Queue。如果业务是“订单已支付事件”多个系统都要收到通常选 Topic。JSON 优先于 Java 原生序列化跨系统消息更适合 JSON。原因可读性更好跨语言更容易版本演进更可控排查消息更方便关键消息设置持久化发送端可以设置jmsTemplate.setDeliveryPersistent(true);原生 JMS 可以设置producer.setDeliveryMode(DeliveryMode.PERSISTENT);还需要确保 Broker 持久化存储配置正常。消费者需要幂等消息可能重复投递。消费者处理逻辑需要支持重复执行。例如订单已处理 - 直接跳过 订单未处理 - 执行业务并记录处理状态失败消息进入 DLQ失败消息不适合无限重试。更常见的做法有限次数重投递 超过次数进入 DLQ 后台补偿或人工处理常用 API 汇总API / 注解作用JmsTemplate.convertAndSend(...)发送消息JmsListener监听消息ConnectionFactory创建连接ConnectionJMS 连接SessionJMS 会话Queue点对点目的地Topic发布订阅目的地MessageProducer消息生产者MessageConsumer消息消费者TextMessage文本消息MappingJackson2MessageConverterJSON 消息转换Session.CLIENT_ACKNOWLEDGE客户端确认session.commit()提交事务会话session.rollback()回滚事务会话RedeliveryPolicy重投递策略总结ActiveMQ是 Java 生态里经典的 JMS 消息中间件。它的核心模型很清晰Queue点对点一条消息一个消费者 Topic发布订阅一条消息多个订阅者日常 Spring Boot 项目里常用组合是spring-boot-starter-activemq JmsTemplate JmsListener MappingJackson2MessageConverter真正落到工程实践需要重点关注Queue 和 Topic 模型选择JSON 消息格式消息持久化事务或确认机制重投递策略DLQ 死信队列消费幂等管理后台监控ActiveMQ 比 RabbitMQ 更贴近 JMS 规范。维护传统 Java 企业系统、对接 JMS 老系统、使用 Queue / Topic 这种经典模型时ActiveMQ 依然有明确价值。