最近项目里有一段业务需要做异步通知我顺手把Java消息服务JMS和Apache ActiveMQ完整过了一遍。之前工作中虽然用过现成的消息中间件但坦白说对JMS规范本身的理解一直停留在“知道个大概”的层面。这次借着ActiveMQ的落地实践从概念到代码把整条链路啃了一遍发现很多以前模糊的地方清晰了不少。这篇算是整理出来的学习笔记写给同样刚接触消息中间件、想系统搞懂JMS和ActiveMQ之间关系的朋友。你看标题里的“ActiveM”其实就是ActiveMQ写的时候手滑少了个Q懒得改了。1. 整体学习思路为什么要从JMS和ActiveMQ入手1.1 消息中间件到底解决了什么问题先把场景摆出来。我之前在一家做电商系统的公司订单模块创建订单之后要通知库存系统减库存、积分系统加积分、推送服务发通知。最早这些逻辑全在订单接口里同步调用服务之间耦合得很死而且响应时间被下游拖得很难看。下单接口平均响应时间800多毫秒大部分时间都耗在等下游返回结果上。后来把调用改成异步订单创建成功后只发一条消息到消息服务器下游系统各自订阅自己关心的消息去消费。接口响应时间直接降到了200毫秒左右系统的吞吐能力也上来了。这就是消息中间件解决的两个核心问题异步解耦和流量削峰。放在更宏观的视角看消息中间件是分布式系统里非常基础的一环不管是订单系统、支付系统还是日志采集系统都会用它来打通数据链路。现在市面上消息中间件产品很多像RabbitMQ、Apache Kafka、RocketMQ都各自有忠实用户。我第一次系统学习时没有直接上Kafka或RocketMQ而是选了ActiveMQ原因有几个ActiveMQ是Apache开源项目部署简单不需要依赖额外的协调组件下载解压就能启动对新手非常友好。它对JMS规范的支持最完整是学习和理解JMS标准的最佳载体。社区资料丰富踩坑时能搜到大量现成的解决方案。在中小规模系统、传统企业内部系统里它的出场率其实比想象中高学会它并不算浪费时间。1.2 先搞懂JMS和ActiveMQ的关系这是新手最容易绕晕的地方。简单说JMSJava Message Service是一套规范是Java世界里定义消息系统访问方式的统一API标准ActiveMQ是实现这套规范的具体消息中间件产品。打个比方JMS就像JDBCActiveMQ就像具体的数据库驱动你通过JDBC接口操作数据库底层连接的是MySQL还是PostgreSQL由驱动决定而ActiveMQ就像是消息领域的“MySQL驱动”加上“数据库本身”。这种关系决定了学习路径先看JMS规范定义的API长什么样再去理解ActiveMQ对规范的具体实现。规范定义了ConnectionFactory、Connection、Session、Destination、MessageProducer、MessageConsumer这些核心接口ActiveMQ则提供了ActiveMQConnectionFactory等具体实现类。你写的业务代码只依赖javax.jms包下的接口换一个JMS实现时代码层面几乎不用改只需要替换连接工厂的配置。这是JMS设计上最聪明的地方。2. JMS核心概念梳理2.1 消息模型点对点与发布订阅JMS规范定义了两种消息模型点对点P2PPoint-to-Point和发布订阅Pub/SubPublish/Subscribe。这两个模型对应两个不同的Destination类型Queue和Topic。点对点模型使用Queue。生产者发送消息到Queue消息服务器保存这条消息然后由某一个消费者消费。关键特征是“一条消息只能被一个消费者消费一次”多个消费者同时监听同一个Queue时消息会在消费者之间竞争分发。消费完之后消息就从Queue中移除。这个模型天然适合任务分发场景比如订单服务发一条“减库存”命令库存服务只有一个消费者实例处理它处理完就完事不会重复处理。发布订阅模型使用Topic。生产者发送消息到Topic所有订阅了这个Topic的消费者都能收到这条消息的副本就像广播。这个模型适合事件通知场景比如用户下单后库存服务、积分服务、短信服务都关心这个“订单创建”事件它们各自订阅同一个Topic各自独立消费。从JMS API角度看创建Queue和Topic的方式几乎一样Queue queue session.createQueue(order.queue); Topic topic session.createTopic(order.topic);2.2 核心API组件逐个说JMS的API由几个核心组件构成逻辑链路非常清晰。ConnectionFactory是连接工厂用来创建Connection。在ActiveMQ中你创建的是ActiveMQConnectionFactory实例构造参数传入broker的连接地址ActiveMQConnectionFactory factory new ActiveMQConnectionFactory(tcp://localhost:61616);Connection代表应用程序到消息服务器之间的物理连接。创建Connection的开销比较大包含了网络连接的建立和认证实际项目中应该复用。Connection创建后需要调用start()方法启动连接否则消费者收不到消息这是个非常隐蔽的坑。Session是JMS中最核心的对象它提供了创建消息、生产者和消费者的方法。Session可以理解为一次“会话上下文”它是单线程使用的。创建Session时可以指定两个参数是否使用事务以及消息确认模式Session session connection.createSession(false, Session.AUTO_ACKNOWLEDGE);Destination是消息目的地就是上面说的Queue或Topic。MessageProducer负责发送消息MessageConsumer负责接收消息。Message是消息本身是生产者与消费者之间传递的数据载体。消息对象分三个部分消息头Header、属性Properties和消息体Body。消息头包含了一些标准的元数据比如JMSDestination、JMSDeliveryMode、JMSMessageID、JMSTimestamp、JMSCorrelationID等。属性部分可以设置自定义的键值对这些属性除了携带业务数据之外还能被用于消息选择器做过滤筛选。消息体是实际的数据内容JMS规范定义了五种消息体类型消息类型消息体的数据形式适用场景TextMessage字符串文本最常用传递JSON、XML数据MapMessage一组键值对key为Stringvalue为Java基本类型传递结构化数据ObjectMessage可序列化的Java对象传递业务对象跨语言时不要用BytesMessage字节数组传递二进制数据如图片、文件StreamMessage基本数据类型的连续流需要按顺序读取的数据流日常业务中TextMessage用的最多因为JSON格式通用性最好跨系统通信时对方不一定是Java。2.3 消息确认机制和事务这块必须吃透JMS可靠性的关键在于确认机制。消费者从broker收到消息后需要在Session层面做“确认”告诉broker这条消息我处理完了你可以移除它了。如果消费者一直不确认broker就会认为消费失败消息会再次投递。确认模式在创建Session时指定一共有几种AUTO_ACKNOWLEDGE自动确认。消费者收到消息后JMS实现自动帮我们确认无需手动干预。这是最简单的模式也是大多数入门示例使用的模式。CLIENT_ACKNOWLEDGE客户端确认。消费者收到消息后必须手动调用message.acknowledge()方法才会确认。如果忘了调用消息可能会被重复投递。DUPS_OK_ACKNOWLEDGE允许重复确认。这种模式下确认是延迟的消息可能被重复投递多次可靠性和性能的折中方案。SESSION_TRANSACTED事务模式。如果Session以事务方式创建需要手动调用session.commit()提交事务消息才会被确认调用session.rollback()则回滚消息重新进入投递队列。事务模式和确认模式是两套机制注意区分。创建Session时第一个boolean参数控制是否启用事务// 非事务、自动确认 Session session connection.createSession(false, Session.AUTO_ACKNOWLEDGE); // 事务模式第二个参数会被忽略 Session session connection.createSession(true, Session.SESSION_TRANSACTED);我在实际项目中踩过一个坑用了CLIENT_ACKNOWLEDGE模式但业务处理逻辑比较复杂偶尔抛出异常后忘记调用acknowledge方法导致消息被重新投递消费端出现重复处理。后来规范了写法把acknowledge调用放在finally块里或者直接统一用事务模式用commit和rollback来兜底。ActiveMQ还扩展了一个INDIVIDUAL_ACKNOWLEDGE模式可以单独确认某一条消息不展开说了知道有这个东西就行。3. 环境搭建与第一个可运行的Demo3.1 下载安装与验证启动我使用的ActiveMQ版本是5.15系列JDK是Java 8。先去Apache官网下载tar.gz或zip包解压到一个目录下。bin目录里有启动脚本Linux系统执行./activemq startWindows系统执行activemq.bat start启动成功后默认监听两个端口61616是OpenWire协议的连接端口客户端通过这个端口连接broker8161是Jetty Web管理控制台的端口浏览器访问http://localhost:8161/默认用户名admin密码admin可以在控制台里看到队列、Topic、连接数等运行状态。验证启动是否正常有几个简便方法。一是用netstat查端口netstat -an | grep 61616二是直接写一个最简的客户端程序去连一下如果连接成功并且能发送消息说明环境没问题。3.2 引入依赖与生产者代码Maven项目引入ActiveMQ客户端依赖dependency groupIdorg.apache.activemq/groupId artifactIdactivemq-all/artifactId version5.15.15/version /dependency注意activemq-all这个依赖会打包ActiveMQ所有相关的类包括broker端和客户端。如果只是写客户端程序也可以用activemq-client这个依赖体积更小但直接使用activemq-all对初学者更省事。下面写一个最简生产者向队列order.queue发送一条文本消息模拟“订单创建成功”事件。import org.apache.activemq.ActiveMQConnectionFactory; import javax.jms.*; public class OrderProducer { public static void main(String[] args) throws Exception { // 1. 创建连接工厂 ActiveMQConnectionFactory factory new ActiveMQConnectionFactory(tcp://localhost:61616); // 2. 创建连接并启动 Connection connection factory.createConnection(); connection.start(); // 3. 创建会话非事务自动确认 Session session connection.createSession(false, Session.AUTO_ACKNOWLEDGE); // 4. 创建目的地 Queue queue session.createQueue(order.queue); // 5. 创建生产者并指定持久化模式 MessageProducer producer session.createProducer(queue); producer.setDeliveryMode(DeliveryMode.PERSISTENT); // 6. 创建消息并发送 String orderJson {\orderId\:\ORDER-20250101-0001\,\amount\:299.00}; TextMessage message session.createTextMessage(orderJson); producer.send(message); System.out.println(订单消息已发送: orderJson); // 7. 释放资源 producer.close(); session.close(); connection.close(); } }代码里有一行容易被忽略但很重要的配置producer.setDeliveryMode(DeliveryMode.PERSISTENT)。持久化消息会写入ActiveMQ的存储默认是kahaDBbroker重启后消息不丢失。非持久化消息只存在于内存中broker一重启就没了。对于订单这类核心业务数据必须使用持久化模式。3.3 消费者同步和异步两种写法消费端有两种主流写法同步receive和异步监听。同步写法使用consumer.receive()方法这个方法会阻塞当前线程直到收到消息或超时import org.apache.activemq.ActiveMQConnectionFactory; import javax.jms.*; public class OrderConsumerSync { public static void main(String[] args) throws Exception { ActiveMQConnectionFactory factory new ActiveMQConnectionFactory(tcp://localhost: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); // 阻塞等待消息设置5秒超时 Message message consumer.receive(5000); if (message instanceof TextMessage) { TextMessage textMessage (TextMessage) message; System.out.println(【同步消费者】收到订单消息: textMessage.getText()); } else if (message null) { System.out.println(【同步消费者】等待消息超时没有收到消息); } consumer.close(); session.close(); connection.close(); } }同步receive适合那种“轮询拉取”的场景比如定时任务从队列里取一批消息去处理。异步写法是设置MessageListener监听器消息到达时由JMS实现自动回调onMessage方法代码不需要主动等待import org.apache.activemq.ActiveMQConnectionFactory; import javax.jms.*; public class OrderConsumerAsync { public static void main(String[] args) throws Exception { ActiveMQConnectionFactory factory new ActiveMQConnectionFactory(tcp://localhost: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); consumer.setMessageListener(new MessageListener() { Override public void onMessage(Message message) { if (message instanceof TextMessage) { try { TextMessage textMessage (TextMessage) message; System.out.println(【异步消费者】收到订单消息: message); // 在这里处理业务逻辑比如减库存、加积分 } catch (JMSException e) { e.printStackTrace(); } } } }); // 异步监听是回调式的main线程不能提前退出做线程阻塞演示 System.out.println(【异步消费者】已启动监听中...); Thread.sleep(60000); consumer.close(); session.close(); connection.close(); } }异步监听有一个注意点MessageListener.onMessage方法在JMS连接的工作线程中执行这个方法返回后才会调用下一轮通知回调。如果你的业务处理时间较长会阻塞后续消息的接收。所以不要在onMessage里做耗时的阻塞操作要拆到独立线程池去处理。3.4 消息选择器按条件过滤消息JMS提供了一种按条件接收消息的机制消息选择器Message Selector。创建消费者时可以传入一个表达式条件只有满足条件的消息才会被投递给这个消费者MapMessage mapMessage session.createMapMessage(); mapMessage.setString(type, order); mapMessage.setDouble(amount, 299.0); producer.send(mapMessage);消费端筛选MessageConsumer consumer session.createConsumer(queue, type order AND amount 100);消息选择器用的是类SQL的语法只能引用消息属性和消息头的标准字段不能引用消息体里的内容。这个功能适合在同一个目的地里区分不同业务场景的消息。4. 实操中踩过的坑和排查技巧4.1 连接被拒绝先查端口再查地址新环境里最容易碰到的问题就是连接时抛出异常核心报错通常是Connection refused。这个问题的排查顺序非常固定用netstat确认61616端口有没有在监听。确认连接的地址对不对如果broker不在本机IP地址一定要写对。如果用了防火墙记得放行61616端口。有一次我发现端口监听正常但客户端就是连不上折腾了很久才发现是配置里把tcp协议写成了http协议。这种低级错误排查时一定要先看连接URL的格式。4.2 消息发了但消费者收不到这个问题的原因通常是连接没有start。Connection对象的start()方法必须在消费者开始接收消息之前调用很多入门代码会忽略这一步结果就是生产者正常发送、broker也收到了消息但消费者端就是没反应因为连接根本没启动。排查思路是去管理控制台看队列的Enqueue和Dequeue计数。如果Enqueue在涨、Dequeue不变说明消息进队列了但消费者没有消费如果消费者代码已部署而且日志里没有异常优先检查connection.start()有没有调用。4.3 确认模式不匹配导致的重复消费用了CLIENT_ACKNOWLEDGE模式但代码里没有调用acknowledge()这是重复消费的典型原因。消息会被broker反复投递直到消费者确认成功。重复消费在业务上会造成很大麻烦比如重复减库存、重复发短信。解决思路有三种一是对生产者和消费者统一使用自动确认模式二是事务Sessioncommit和rollback强制管理三是业务幂等处理通过唯一业务键去重。实际项目中建议至少做好第三点因为分布式场景下消息超时重投、网络抖动导致的重复其实是常态。4.4 消息丢失持久化与非持久化的区别默认情况下如果生产者在send之前没有设置持久化模式消息是非持久化的broker重启之后消息就丢了。很多人在测试阶段没注意这一点到了生产环境才发现broker一重启队列里的数据全部清空。ActiveMQ的持久化机制默认使用kahaDB一种基于文件的内存磁盘混合存储它的性能可以满足大多数中小系统的要求。如果需要更高可靠性可以切换为JDBC持久化把消息存储到数据库表里但写入性能会下降要谨慎评估。对核心业务消息务必显式设置deliveryMode为PERSISTENT。对实时性要求高、丢失影响小的数据如实时日志用非持久化消息性能更好。4.5 管理控制台的几个重要指标ActiveMQ管理控制台的queue页面有几个指标要看懂Number Of Pending Messages队列中等待被消费的消息数量。这个值如果持续增长说明消费速度跟不上生产速度。Number Of Consumers当前消费者的数量。如果为0说明没有消费者在线消息会一直积压。Messages Enqueued累计进入队列的消息总数。Messages Dequeued累计被消费掉的消息总数。例行检查时重点看第一个指标正常情况下它应该在一个低位水平波动。如果持续上涨就要看消费者是不是挂了、消费速度是不是太慢或者下游出问题导致消费者反复异常。4.6 关于connection.start()和close()的小规矩JMS资源释放顺序是有讲究的关闭顺序必须是producer/consumer先关再关session最后关connection。如果反着来有时会出现连接已经关闭但消息还在发送中的间歇性异常。还有一个小经验connection的创建开销很大每次发送消息都新建连接再关闭性能极差正确做法是复用一个Connection和Session把producer做成轻量复用对象。当然这只是入门阶段知道方向就好真到了高并发阶段就该上连接池了。5. 最后再分享一点个人的学习体会这次系统学习JMS和ActiveMQ最大的收获不是学会了一个具体产品的API而是理解了消息系统的通用模型。后来再看RabbitMQ和Kafka的文档发现很多概念是相通的比如持久化、确认机制、消费者组、消息堆积底层逻辑都一样。ActiveMQ里这些概念用最直白的方式呈现作为入门教材不可替代。我觉得生产上使用ActiveMQ要特别注意的就是确认机制和持久化这两个点最好一开始就养成显式配置的好习惯。很多线上P0级别的消息问题不是ActiveMQ本身出了故障而是使用方没有理解这两套机制在错误的地方用了默认值。这篇笔记到这里基本就结束了如果你也在入门消息中间件希望这篇内容能帮你少踩一些我踩过的坑。