做消息中间件选型的时候很多人第一反应是上大厂那套重家伙好像不搞个分布式集群就没法跟人打招呼。我这次倒是反着来因为一个积分系统要异步接入订单系统订单高峰期HTTP回调经常超时重试又导致下游重复记账最后只能老老实实把JMS和ActiveMQ从概念到落地完整过了一遍。学完之后有个很深的体会——JMS作为Java世界的消息规范和ActiveMQ这个具体实现的关系特别像JDBC和MySQL的关系。搞懂这层关系后面写代码、排问题都会顺很多。这篇文章就是把这趟学习过程整理出来适合刚接触消息队列、准备在项目里落地ActiveMQ的开发者。我会从“为什么需要它”讲到“API模型怎么理解”再到“安装部署、双端通信代码、可靠性和事务细节”最后把实际排查中遇到的几个坑完整复盘一遍。内容以实践为主不堆概念。1. 一段真实业务场景为什么最终选了JMS ActiveMQ1.1 积分服务和订单服务之间的“重试风暴”先说业务背景。模拟项目里有一个订单服务和一个积分服务订单创建成功后要调用积分服务给用户加积分。最开始直接走HTTP同步调用逻辑很简单订单落库调积分接口返回成功就算完。坏就坏在高峰期——积分服务偶尔抖动一个请求超时调用方就重试重试又叠加上来积分服务被打得更慢形成恶性循环。我翻日志的时候见过同一笔订单在5分钟内被重试了17次。同步调用是典型的“强耦合、无缓冲”上游要等下游返回结果下游一慢上游整个请求链路都拖慢。更麻烦的是订单服务多实例部署时每个实例都在重试流量被放大了好几倍。这时候需要的其实不是更强的服务器而是一个“缓冲层”——订单服务把消息发出去就返回积分服务按自己的速度去消费重试和削峰的事情交给消息中间件处理。1.2 为什么要按JMS规范选型而不是直接绑定实现选型的时候我对比了几个Java生态里常见的消息中间件。Kafka偏向日志流、高吞吐场景但部署和运维成本高另外两个也都不错但项目的规模实在不需要那么重的集群。最终选ActiveMQ其中一个关键原因是它完整实现了JMS规范也就是说你写的生产者和消费者代码依赖的是javax.jms包下的标准接口。哪天团队觉得ActiveMQ撑不住了换成其他兼容JMS规范的产品业务代码改动量会小很多。这跟JDBC一个道理应用层写java.sql.Connection底层换成MySQL还是别的数据库驱动换掉就差不多了。抽象层最大的价值不是让代码少写几行而是把“业务逻辑”和“具体中间件”之间做一个隔离出问题的时候有一个清晰的替换边界。另一个原因是ActiveMQ部署足够轻单机解压就能跑功能上队列、主题、持久化、事务、消息过滤都有非常适合中小规模项目。1.3 什么场景不该硬上ActiveMQ学习过程中我也梳理了反面清单。如果项目就是两个服务之间的简单通知一天几百条调用同步接口完全能扛住引入消息队列反而多出一个需要维护的组件。如果业务要求强实时、强一致比如支付成功后要立刻拿到结果展示给用户那MQ的异步特性就不合适还是要走同步接口加事务。我的建议是先想清楚问题是不是“削峰填谷、解耦、异步化”这三类。如果是MQ是合适的如果不是别为了技术栈好看而上消息中间件。表格总结一下场景特征是否适合引入ActiveMQ原因接口间需要异步解耦下游处理慢适合上游无需等待削峰高峰期流量集中存在瞬时压力适合队列缓冲控制消费速率广播通知、事件分发适合Topic模型天然支持发布订阅调用量极小同步两三百毫秒内完成不适合增加维护成本和链路复杂度要求立即拿到最终结果不适合异步无法保证实时返回2. JMS核心对象模型从Connection到Message的完整链条2.1 JMS和ActiveMQ的“规范与实现”双层结构JMS全称Java Message Service是Java平台上定义的一套消息服务标准接口规定了生产者和消费者怎么创建连接、怎么收发消息、消息格式长什么样。但这些接口本身不能干活真正干活的是像ActiveMQ这样的消息中间件它实现了这套接口并提供底层的Broker服务。初学者最容易绕晕的就是这里代码里import的全是javax.jms开头的包好像没直接用到ActiveMQ的类但跑起来链路明明就是通过ActiveMQ的Broker在传输消息。其实工厂类ActiveMQConnectionFactory是ActiveMQ自己实现的它构造出来的Connection、Session、MessageProducer等对象都是在JMS标准接口背后做了具体实现的。所以学习的时候我建议先把JMS的接口模型搞清楚再去看ActiveMQ的配置和参数就不会被代码里的包名迷惑了。2.2 对象模型的完整链条和生命周期一条完整的好消息链路大致是这个顺序先创建连接工厂由工厂创建连接连接创建会话会话再创建消息生产者和消费者然后通过Destination目的地来指定消息发到哪里。用生活里的类比就是ConnectionFactory是快递公司的加盟网点Connection是快递员Session是快递员手里的工作单Destination是收货地址MessageProducer是把包裹交给快递员的动作MessageConsumer是收件人签收的动作。有一个特别容易踩的坑是Session的使用边界。JMS规范里Session不是线程安全的一个Session建议只在一个线程里使用。很多新手把Session定义成全局单例然后开一堆线程同时往里塞消息结果就出现偶发性的消息丢失或者报错。后面踩坑章节我会专门复盘这个问题。另一个要注意的是Connection和Session都要有明确的关闭动作顺序上先关生产者、消费者再关Session最后关Connection尽量在finally块里执行。2.3 消息的几种类型到底该怎么选JMS定义了五种消息类型TextMessage、MapMessage、BytesMessage、ObjectMessage、StreamMessage。我自己的使用习惯是这样的TextMessage最常用传JSON字符串非常舒服前后端、跨语言都方便。MapMessage适合传结构化键值对但键类型受限灵活性一般。BytesMessage适合传二进制内容比如文件分片、图片数据。ObjectMessage理论上能传Java对象但我基本不推荐序列化和版本兼容问题多。StreamMessage是流式读写实际项目用得少。类型适合场景注意点TextMessageJSON、XML、纯文本最通用MapMessage固定键值结构键值受限BytesMessage二进制文件需自己处理字节流ObjectMessageJava对象传输类版本变化会影响反序列化StreamMessage流式原语使用较少文本消息的好处是后面做消息格式升级方便消费方拿到字符串自己解析兼容老格式就是代码里多几个判断的问题。换成ObjectMessage老版本对象反序列化失败那就是事故了。2.4 Queue与Topic点对点和发布订阅的关键差异Queue对应点对点模型Point-to-Point一条消息只会被一个消费者消费。哪怕你有十个消费者在监听同一个Queue这条消息也只会发给其中一个。消费完成后消息就被移出队列不存在“群发”的效果。Topic对应发布订阅模型Pub/Sub生产者发布一条消息所有订阅了这个Topic的消费者都能收到。初学者经常在这一步栽跟头用Topic就是想做广播但广播要求消费者必须保持在线订阅状态离线期间的Topic消息是收不到的除非配了持久订阅。Queue则相反消息只要到达Broker就会持久留存配置持久化后消费者后启动也能补收。理解这个差异后面很多“消息不见了”的问题都能避免。3. ActiveMQ落地安装、启动与第一个消息收发3.1 本地环境快速跑起一个Broker实例ActiveMQ的安装没有太多花活官网下载压缩包解压后进到bin目录Windows下运行activemq.bat startLinux下运行./activemq start默认就起来了。启动后默认监听61616端口这是OpenWire协议端口客户端连接主要走这个端口。另一个是8161端口这是Web控制台端口浏览器打开能看消息的收发量、消费者状态还有简单的消息浏览功能。我第一次启动的时候卡了一个小问题控制台打不开后来发现是防火墙把8161端口挡了。还有一次是启动脚本提示端口被占用用lsof -i:61616查了一下发现之前有个残留的Java进程没退干净。如果是第一次玩建议先用默认配置跑通熟悉了再改conf/activemq.xml。控制台默认账号通常是admin/admin但生产环境要第一时间改掉这个后面再说。3.2 Maven依赖和最小生产者代码新建一个工程加两个依赖就够dependency groupIdorg.apache.activemq/groupId artifactIdactivemq-client/artifactId version5.18.2/version /dependency dependency groupIdorg.apache.activemq/groupId artifactIdactivemq-pool/artifactId version5.18.2/version /dependencyactivemq-client是核心客户端库activemq-pool后面做连接池会用。生产者的代码其实很短public class SimpleProducer { public static void main(String[] args) throws JMSException { // 1. 创建连接工厂Broker地址为本地 ActiveMQConnectionFactory factory new ActiveMQConnectionFactory( admin, admin, tcp://localhost:61616); // 2. 创建连接并启动 Connection connection factory.createConnection(); connection.start(); // 3. 创建会话第一个参数表示是否开启事务这里用false Session session connection.createSession(false, Session.AUTO_ACKNOWLEDGE); // 4. 创建消息目的地 Queue queue session.createQueue(test.queue); // 5. 创建生产者 MessageProducer producer session.createProducer(queue); // 6. 发送消息 TextMessage message session.createTextMessage(hello activemq); producer.send(message); System.out.println(消息发送完成); // 关闭资源 producer.close(); session.close(); connection.close(); } }这个流程就是JMS标准链路的标准写法。注意connection.start()有些同学会漏掉结果生产者能发但消费者收不到消息其实是因为连接没启动导致消费者无法接收。3.3 最小消费者代码同步接收和异步监听消费者的写法有两种。一种是主动拉取调用receive()方法阻塞等待消息另一种是注册一个监听器Broker有消息来了主动回调。先看同步接收public class SimpleConsumer { public static void main(String[] args) throws JMSException { ActiveMQConnectionFactory factory new ActiveMQConnectionFactory( admin, admin, tcp://localhost:61616); Connection connection factory.createConnection(); connection.start(); Session session connection.createSession(false, Session.AUTO_ACKNOWLEDGE); Queue queue session.createQueue(test.queue); MessageConsumer consumer session.createConsumer(queue); // 阻塞接收设置超时时间 Message message consumer.receive(1000); if (message instanceof TextMessage) { TextMessage textMessage (TextMessage) message; System.out.println(收到消息: textMessage.getText()); } consumer.close(); session.close(); connection.close(); } }同步接收适合需要控制节奏的场景比如批量拉取任务。异步监听适合事件通知类业务consumer.setMessageListener(new MessageListener() { Override public void onMessage(Message message) { if (message instanceof TextMessage) { try { TextMessage textMessage (TextMessage) message; System.out.println(异步收到消息: textMessage.getText()); } catch (JMSException e) { // 这里的异常处理非常关键 e.printStackTrace(); } } } });异步监听里最忌讳的就是onMessage方法里抛异常后什么都不处理。默认情况下AUTO_ACKNOWLEDGE模式会在onMessage正常返回后自动确认消息如果抛了异常且未确认消息会被重新投递极容易造成无限重发循环。这个我放到踩坑部分详细说。3.4 用控制台验证消息是否正常跑完上面两段代码后打开8161控制台找到对应的Queue。有两个计数值得关注Enqueued表示进入队列的消息总数Dequeued表示消费完成移出队列的消息数。两者相等说明消息都消费完了。如果Dequeued远小于Enqueued说明有消息积压或者消费者没消费。这个页面就是后面排查问题时的第一现场。4. 双端通信实战连接池、failover与JmsTemplate封装4.1 为什么不能用裸Connection到处创建前面示例为了方便每个消息都直接new一个Connection这在demo里没问题但真实项目里不能这么干。建立一条连接底层要创建TCP连接、完成认证握手、创建传输通道成本不低。如果每次发消息都现场创建、用完关闭高并发下Broker会收到大量无意义的建连请求消费端连接数也容易被打满。解决办法就是像数据库连接池一样引入ActiveMQ连接池。activemq-pool提供了PooledConnectionFactory可以复用连接和会话。配置了几个关键参数之后整个资源管理就自动了用完归还不频繁建连。ActiveMQConnectionFactory factory new ActiveMQConnectionFactory( tcp://localhost:61616); // 连接池工厂 PooledConnectionFactory pooledFactory new PooledConnectionFactory(factory); // 池中最多多少个连接 pooledFactory.setMaxConnections(10); // 每个连接上最多活跃的会话数 pooledFactory.setMaximumActiveSessionPerConnection(100); pooledFactory.start();这里的maxConnections和数据库连接池的maximumPoolSize一个意思。可以根据业务并发量估算平均延迟50毫秒单连接撑住的并发QPS大概也就几十要支撑更高的并发就要多几条连接来分流。4.2 failover协议连接断了怎么办单台Broker如果挂了客户端连接会直接断掉生产环境必须有高可用手段。ActiveMQ的客户端支持failover协议连接串写成这样failover:(tcp://192.168.1.10:61616,tcp://192.168.1.11:61616)?randomizetrueinitialReconnectDelay100maxReconnectAttempts10多个地址之间用逗号分隔一旦当前节点不可用客户端会自动重连到另一个节点。randomizetrue表示客户端随机选择初始连接节点避免所有客户端都连到同一台机器上实现简单的负载分散。initialReconnectDelay是第一次重连的延迟毫秒数maxReconnectAttempts是最大重连次数超过后就会抛出异常。这个方案适合主从、集群部署的场景。当然failover只是客户端层面的自动重连Broker之间的数据同步还要靠后端的网络配置和持久化机制不能只靠一个URL就解决所有高可用问题。4.3 封装一个通用消息发送类实际项目中消息发送一般会统一封装成一个组件避免每个业务方都去操作底层JMS API。我用JmsTemplate做过一个简单的封装Component public class JmsMessageSender { private final JmsTemplate jmsTemplate; public JmsMessageSender(ConnectionFactory connectionFactory) { this.jmsTemplate new JmsTemplate(connectionFactory); this.jmsTemplate.setDeliveryMode(DeliveryMode.PERSISTENT); } public void sendQueue(String queueName, String message) { jmsTemplate.send(queueName, s - s.createTextMessage(message)); } public void sendTopic(String topicName, String message) { jmsTemplate.send(topicName, s - s.createTextMessage(message)); } public void sendWithTimeout(String queueName, String message, long timeToLive) { jmsTemplate.setTimeToLive(timeToLive); jmsTemplate.convertAndSend(queueName, message); } }这个类将“目的地名称”和“业务消息内容”隔离开业务方只需要调用方法传入队列名和消息体就行至于连接从哪里来、会话怎么管理、消息确认怎么设置全部由模板完成。DeliveryMode.PERSISTENT表示消息要持久化Broker收到后会先落盘再返回成功保证Broker重启后消息不丢。代价是性能会差一些适合重要业务如果是日志类可容忍丢失的数据可以改成NON_PERSISTENT换取更高的吞吐。4.4 请求-响应模式TemporaryQueue的玩法除了发完就完事的消息JMS还支持同步请求-响应模式。客户端发送消息后可以通过临时队列等待服务端处理结果。临时队列是Connection级别独立的消费端可以获取到请求消息里的JMSReplyTo字段把结果发回这个临时队列。// 生产端 Queue replyQueue session.createTemporaryQueue(); MessageProducer producer session.createProducer(requestQueue); TextMessage textMessage session.createTextMessage(查询库存); textMessage.setJMSReplyTo(replyQueue); producer.send(textMessage); MessageConsumer consumer session.createConsumer(replyQueue); Message reply consumer.receive(3000); if (reply ! null) { System.out.println(响应内容: ((TextMessage) reply).getText()); }这是个很有用的模式比如调用一个慢服务时不想同步等待太长时间就把请求放到消息队列里让服务端异步处理完再通过临时队列回传结果。但要注意设置超时时间万一服务端挂了客户端不能被永久卡死。临时队列还有个特点Connection一断它就被销毁所以这个模式不太适合Broker故障恢复的场景使用前要权衡好。5. 可靠性细节持久化、事务、确认机制与死信处理5.1 Broker端持久化存储kahadb、jdbc store怎么选ActiveMQ的持久化主要解决“Broker重启之后消息还在不在”的问题。默认用的是kahadb这是一个基于文件追加日志的存储引擎性能不错吞吐量高适合绝大多数场景。配置在activemq.xml里默认就会启用。如果消息量特别大也可以改成jdbc store把消息存进数据库好处是可以用数据库的备份、恢复方案坏处是性能通常比文件型存储低不少。从可靠性的角度我建议把这几个层次理解清楚存储类型特点适合场景kahadb默认文件型性能好通用jdbc store落库管理方便性能略差已有数据库运维体系的团队amq store老版文件存储不推荐新项目使用持久化不只是Broker端的事消息本身也要标记为持久化。DeliveryMode.PERSISTENT配合Broker持久化才能保证消息在Broker重启后不丢。如果哪一层没做好重启后消息就可能消失。5.2 事务模式生产者事务与消费者事务Session的第一个参数是transacted设置为true就表示开启事务。生产端开启事务后可以连续发送多条消息最后调用session.commit()统一提交。只有提交成功消息才会真正入队。如果中途出错调用session.rollback()这批消息全部撤销。消费端的事务和确认机制挂钩。消费端开启事务后消息处理完成需要commit()消息才算被确认消费处理失败可以rollback()消息会被重新投递。我写消费端代码的习惯是Session session connection.createSession(true, Session.SESSION_TRANSACTED); try { Message message consumer.receive(1000); // 处理业务逻辑 processBusiness(message); session.commit(); // 处理成功确认提交 } catch (Exception e) { session.rollback(); // 处理失败回滚重投 throw e; }事务很有用但也别把整个业务逻辑塞进来。事务范围越大锁和冲突的可能性越大。这个道理和数据库事务是一模一样的要短、要快、要明确边界。5.3 四种确认模式各自解决什么问题说到消费端确认模式JMS提供了四类但都是围绕“什么时候告诉Broker可以删掉这条消息”这一件事。确认模式时机风险AUTO_ACKNOWLEDGE消息接收成功或回调返回时自动确认可能出现接收后、处理前崩溃导致消息丢失CLIENT_ACKNOWLEDGE代码里手动调用message.acknowledge()忘记调用就永远不确认消息积压DUPS_OK_ACKNOWLEDGE允许重复确认性能高可能重复消费SESSION_TRANSACTED配合事务Sessioncommit才算确认rollback重新投递我的建议是核心业务尽量用CLIENT_ACKNOWLEDGE或者事务模式。因为AUTO_ACKNOWLEDGE的确认时机是在消息被消费方接收后立刻确认如果那一瞬间消费方宕机消息已出队列但业务还没处理完重启后这条消息就彻底丢了。手动确认可以把“确认”延迟到业务处理完之后牺牲一点复杂度换取可靠性。5.4 死信队列和RedeliveryPolicy一条消息最多重试几次ActiveMQ默认有一条叫ActiveMQ.DLQ的死信队列。消息消费失败、需要重新投递的次数超过阈值后会被移入死信队列不再骚扰正常消费流程。默认情况下一条消息最多重投6次。这个次数可以配置policyEntry queue redeliveryPolicy property namemaximumRedeliveries value3/ property nameinitialRedeliveryDelay value1000/ /redeliveryPolicy /policyEntryinitialRedeliveryDelay是延迟多久再投递避免失败后马上重投把无效请求打爆。死信队列的价值在于它把“失败的终究会失败”的消息隔离出来运维人员可以去查为什么失败、能不能人工修复后重新处理而不是让系统无限重试。5.5 消费端幂等消息重复是一种常态讲了这么多可靠性机制我必须强调一点在高可用体系里消息重复消费比消息丢失更常见、也更难缠。比如消费端处理了消息、业务写库成功但ack超时了Broker以为没消费又把同一笔消息重新投递。如果消费端的接口不是幂等的库存就扣了两次账就对不上了。最简单的解法是给业务消息加一个全局唯一消息ID消费端用一个去重表记录已经处理过的ID遇到重复消息直接跳过。消息ID可以用订单号、业务主键、或者UUID只要能保证全局唯一即可。这个设计不复杂但能避免非常多的对账噩梦值得在项目初期就做进去。6. 踩坑实录消息丢失、重复消费与堆积排障链路6.1 坑一消息发完之后“消失”了问题出在Topic订阅方式现象生产者是先启动的发了一条Topic消息日志显示发送成功。之后消费者启动并订阅了这个Topic结果一条消息都收不到。排查过程先到控制台看Topic的Enqueued计数确认消息确实进入了Topic。然后排查消费者订阅方式发现用的是非持久订阅。这就命中了我前面说的点订阅者在线时才收得到消息离线期间发布的Topic消息直接被丢弃。如果希望离线订阅者也能收到有两个方案一是权衡业务场景改用Queue二是用持久订阅DurableSubscription。Topic和Queue最大的区别就在这里千万别拿Queue的思维套到Topic上。最终结论这不是ActiveMQ的问题是模型理解错位。把场景改成事件通知时要确认是否有消费者离线补偿的需求。如果有用Queue或者持久订阅如果没有比如是一个实时行情通知错过就算了用普通Topic即可。6.2 坑二消费端莫名重复消费问题出在AUTO确认和长耗时处理现象明明业务已经处理完成但日志里看到了同一条消息被消费了两次。排查了很久找不到原因。排查链路查看消费端处理逻辑发现onMessage里处理了一条耗时较长的业务比如调外部接口要3秒。而这时的确认模式是AUTO_ACKNOWLEDGE消息在接收时就被预确认了如果处理时间过长Session与Broker之间的ack状态出问题Broker可能认为消息未被消费而重新投递。另外如果prefetch消息数量不为1消费端会一次性拉取多条处理完时统一确认这中间消费端一旦宕机整批消息全部重新投递重复消费的范围会更大。解决方法分两步确认模式改为CLIENT_ACKNOWLEDGE处理完业务后手动调用acknowledge()同时把消费端的prefetchSize调成1让每条消息独立确认。ActiveMQPrefetchPolicy prefetchPolicy new ActiveMQPrefetchPolicy(); prefetchPolicy.setQueuePrefetch(1); factory.setPrefetchPolicy(prefetchPolicy);当然最稳妥的仍是幂等设计因为无论如何都有极小的概率重复投递代码里做一层去重才能彻底兜底。6.3 坑三队列堆积告警慢消费者把Broker内存吃满现象某个Queue的Pending消息数从几百涨到几十万消费速率远低于生产速率控制台显示Broker内存使用率持续升高其他队列的消费也收到影响。排查过程先看消费端日志发现单个消息的处理时间平均从50毫秒涨到了500毫秒以上原因是有个查询走了全表扫描数据库慢查询拖累了主流程。这是典型的慢消费者问题。消费者的消费速率不匹配生产速率消息堆积是必然结果。这时候单纯加消费者实例的确能提升并行消费能力但根本性问题在于处理链路本身太慢。另一个隐藏原因是默认prefetch值过大。ActiveMQ默认的队列预取数是1000也就是消费者一次拉取1000条保存在本地内存里慢慢处理。如果消费者内存配得不大这1000条就可能撑爆堆内存。对策是减小prefetch值类似setQueuePrefetch(10)同时保证消费端处理逻辑快速退出不要让一条消息的处理时间无限制拉长。这次排查给我的教训队列积压最先怀疑的永远是消费端不是Broker。控制台看一眼Dequeued速率就知道问题了别一开始就怀疑中间件配置。6.4 坑四多线程共享同一个Session偶发性JMSException现象项目里为了加速消费一个消费者组里开了多个线程共用同一个Session去创建Consumer运行一段时间后偶尔抛JMSException提示Session正忙或已关闭。排查链路查JMS规范里面明确写了Session不是线程安全的。一个Session就要在一个专用线程里使用多线程共享时底层OpenWire协议发送的packet顺序会乱就会出现偶发报错。解决方案是每个消费者线程创建独立Session或者用连接池的PooledSession来替代。用连接池之后每次从池里借一个Session用完了归还天然规避了这个问题。这一点其实是最容易被忽视的JMS的接口看起来很安全实际上坑藏在并发模型里。任何要共享的资源都要先问一句文档里有没有说明线程安全没有说明的默认当成不安全处理。最后说一句我个人的体会。学ActiveMQ如果只看文档很容易记了一堆名词却不知道怎么用真正有效的做法是先把JMS的接口模型画一遍然后拿一个最简单的收发demo跑通再往里面加连接池、事务、持久化这些参数。每加一个特性都要到控制台或者日志里验证它的影响是什么。遇到问题时别急着改配置先顺着消息流转的链路走一遍往往很快就能定位到“模型理解错位”还是“资源使用方式不对”。整个过程其实不复杂但确实需要耐心。