ActiveMQ消息中间件：Java消息通信实战指南

1. ActiveMQ与Java消息通信基础ActiveMQ作为Apache旗下的开源消息中间件实现了JMS 1.1规范为Java应用提供了可靠的消息通信能力。在实际项目中我们经常需要实现不同模块间的解耦通信这时ActiveMQ的队列Queue和主题Topic模式就能派上用场。1.1 环境准备与依赖配置首先需要准备ActiveMQ服务端可以从官网下载最新版本。安装过程非常简单解压后进入bin目录根据操作系统选择对应的启动脚本Windows:activemq.batLinux/Mac:./activemq start验证服务是否启动成功可以访问管理控制台http://localhost:8161/admin默认账号密码都是admin。Java项目中使用Maven添加依赖dependency groupIdorg.apache.activemq/groupId artifactIdactivemq-all/artifactId version5.16.3/version /dependency注意生产环境建议使用连接池可以添加org.apache.activemq:activemq-pool依赖来获取连接池支持。1.2 基础通信模型实现消息生产者的基本实现流程// 1. 创建连接工厂 ConnectionFactory factory new ActiveMQConnectionFactory(tcp://localhost:61616); // 2. 创建连接并启动 Connection connection factory.createConnection(); connection.start(); // 3. 创建会话事务性会话自动确认 Session session connection.createSession(true, Session.AUTO_ACKNOWLEDGE); // 4. 创建队列目的地 Destination queue session.createQueue(TEST.QUEUE); // 5. 创建生产者并发送消息 MessageProducer producer session.createProducer(queue); TextMessage message session.createTextMessage(Hello ActiveMQ); producer.send(message); // 6. 提交事务并关闭连接 session.commit(); connection.close();对应的消费者实现// 1-4步与生产者相同 MessageConsumer consumer session.createConsumer(queue); // 5. 接收消息同步阻塞方式 Message msg consumer.receive(5000); // 超时5秒 if(msg instanceof TextMessage) { System.out.println(((TextMessage)msg).getText()); } // 6. 同样需要提交和关闭 session.commit(); connection.close();2. 高级特性与实战技巧2.1 消息确认机制详解ActiveMQ支持多种消息确认模式通过Session创建时的第二个参数指定Session.AUTO_ACKNOWLEDGE自动确认默认Session.CLIENT_ACKNOWLEDGE客户端手动确认Session.DUPS_OK_ACKNOWLEDGE延迟确认Session.SESSION_TRANSACTED事务会话手动确认示例Session session connection.createSession(false, Session.CLIENT_ACKNOWLEDGE); // ... Message msg consumer.receive(); msg.acknowledge(); // 显式确认经验生产环境中建议使用事务会话(SESSION_TRANSACTED)可以确保消息的可靠传输。我曾遇到过因网络闪断导致消息丢失的情况改用事务后问题解决。2.2 消息监听器模式相比阻塞式的receive()方法更推荐使用消息监听器实现异步处理consumer.setMessageListener(new MessageListener() { Override public void onMessage(Message message) { try { // 处理消息 System.out.println(Received: ((TextMessage)message).getText()); // 模拟业务处理 Thread.sleep(1000); } catch (Exception e) { e.printStackTrace(); } } }); // 需要保持连接不关闭2.3 消息持久化与非持久化通过设置消息的传递模式来决定是否持久化// 持久化消息默认 producer.setDeliveryMode(DeliveryMode.PERSISTENT); // 非持久化消息 producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);持久化消息会被写入磁盘确保Broker重启后不丢失但性能较低非持久化消息只存在内存中性能更高但不可靠。3. Spring整合实战3.1 Spring JMS配置Spring Boot中配置ActiveMQ非常简单# application.yml spring: activemq: broker-url: tcp://localhost:61616 user: admin password: admin packages: trust-all: true # 信任所有包配置JmsTemplateConfiguration EnableJms public class JmsConfig { Bean public JmsTemplate jmsTemplate(ConnectionFactory connectionFactory) { JmsTemplate template new JmsTemplate(connectionFactory); template.setDeliveryMode(DeliveryMode.PERSISTENT); return template; } }3.2 消息发送与接收使用JmsTemplate发送消息Autowired private JmsTemplate jmsTemplate; public void sendMessage(String destination, String message) { jmsTemplate.convertAndSend(destination, message); // 或者使用MessageCreator jmsTemplate.send(destination, session - { return session.createTextMessage(message); }); }接收消息public String receiveMessage(String destination) { return (String) jmsTemplate.receiveAndConvert(destination); }3.3 注解式监听器Spring提供了更简洁的监听器定义方式JmsListener(destination TEST.QUEUE) public void handleMessage(String message) { System.out.println(Received: message); // 业务处理 }可以配置并发消费者JmsListener(destination TEST.QUEUE, concurrency 5-10) public void handleMessage(String message) { // 处理逻辑 }4. 生产环境注意事项4.1 性能调优建议连接池配置Bean public ConnectionFactory connectionFactory() { PooledConnectionFactory pool new PooledConnectionFactory(); pool.setConnectionFactory(new ActiveMQConnectionFactory(tcp://localhost:61616)); pool.setMaxConnections(100); pool.setMaximumActiveSessionPerConnection(50); return pool; }消息压缩对于大消息可以启用压缩ActiveMQConnectionFactory factory new ActiveMQConnectionFactory(); factory.setUseCompression(true);预取策略调整消费者预取数量ActiveMQConnectionFactory factory new ActiveMQConnectionFactory(); factory.getPrefetchPolicy().setQueuePrefetch(100);4.2 常见问题排查消息堆积监控队列深度设置合理的TTL// 设置消息过期时间毫秒 producer.setTimeToLive(60000);内存溢出配置内存限制policyEntry queue memoryLimit100mb/连接断开配置失败重连factory.setOptimizeAcknowledge(true); factory.setAlwaysSessionAsync(false);4.3 监控与管理通过JMX监控ActiveMQ-Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.port1099 -Dcom.sun.management.jmxremote.sslfalse -Dcom.sun.management.jmxremote.authenticatefalse或者使用Hawtio等可视化工具进行监控。我曾通过监控发现某个队列消费速度异常最终定位到是消费者逻辑中存在死锁问题。5. 扩展应用场景5.1 延迟消息实现ActiveMQ支持延迟消息投递MessageProducer producer session.createProducer(queue); TextMessage message session.createTextMessage(Delayed message); message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, 60000); // 延迟60秒 producer.send(message);需要Broker开启调度支持broker xmlnshttp://activemq.apache.org/schema/core schedulerSupporttrue5.2 消息选择器使用JMS选择器过滤消息// 消费者只接收priority高的消息 MessageConsumer consumer session.createConsumer(queue, priority high);生产者设置消息属性message.setStringProperty(priority, high);5.3 集群部署方案ActiveMQ支持多种集群方式Network of Brokers多个Broker组成网络networkConnectors networkConnector uristatic:(tcp://remotehost:61616)/ /networkConnectorsMaster-Slave主从复制提供高可用共享存储使用共享数据库或文件系统在实际项目中我们采用Network of Brokers方案实现了跨机房消息路由配合镜像队列确保了消息的高可靠性。