深入浅出ActiveMQ：消息中间件的核心实践与拓展

咦咦咦，各位小可爱，我是你们的好伙伴——bug菌，今天又来给大家普及Java SE相关知识点了，别躲起来啊，听我讲干货还不快点赞，赞多了我就有动力讲得更嗨啦！所以呀，养成先点赞后阅读的好习惯，别被干货淹没了哦~

🏆本文收录于「滚雪球学Java」专栏中，这个专栏专为有志于提升Java技能的你打造，覆盖Java编程的方方面面，助你从零基础到掌握Java开发的精髓。赶紧关注，收藏，学习吧！

环境说明：Windows 10 + IntelliJ IDEA 2021.3.2 + Jdk 1.8

前言

在现代分布式系统和微服务架构中，消息中间件扮演着至关重要的角色。它不仅解决了系统间的解耦和异步通信问题，还提高了系统的可扩展性和可靠性。其中，ActiveMQ作为一款成熟的开源消息中间件，以其高性能和丰富的功能在业界广受赞誉。本文将全面深入地探讨ActiveMQ的核心原理、实践应用，以及在不同场景下的拓展使用，旨在帮助读者深入理解并灵活运用ActiveMQ，加速企业应用的开发与部署。

一、ActiveMQ概述

1.1 什么是ActiveMQ

ActiveMQ是由Apache基金会开发的一款功能强大的开源消息代理（Message Broker），遵循JMS（Java Message Service）规范。它支持多种消息传输协议，如AMQP、MQTT、STOMP等，能够满足不同场景下的消息传输需求。ActiveMQ具有高性能、可靠性和可伸缩性，适用于各种规模的企业级应用。

1.2 ActiveMQ的核心特点

• 多协议支持：兼容多种消息传输协议，方便跨语言、跨平台的系统集成。
• 高可用性：支持集群部署、主从复制和故障转移，确保系统的连续性。
• 丰富的功能：提供消息持久化、事务支持、安全认证、消息过滤等高级特性。
• 易于管理：提供友好的Web管理界面和多种监控工具，便于系统运维。
• 可扩展性强：通过插件机制和开放的API，可根据业务需求扩展功能。

1.3 消息中间件的作用

• 解耦：发送方和接收方无需同时在线，降低系统间的耦合度。
• 异步通信：支持消息的异步传递，提高系统的响应速度。
• 流量削峰：通过消息队列，平滑处理高并发请求，防止系统过载。
• 可靠传输：确保消息在网络故障或系统异常情况下不丢失。

二、ActiveMQ的核心原理

2.1 JMS模型解析

ActiveMQ基于JMS规范，实现了**点对点（Point-to-Point）和发布/订阅（Publish/Subscribe）**两种消息模型。

2.1.1 点对点模式

• 消息队列（Queue）：消息被发送到特定的队列，消费者从队列中读取消息。
• 特点：每个消息只被一个消费者消费，消息被消费后即被移除。
• 应用场景：需要保证每个消息只处理一次的场景，如订单处理。

2.1.2 发布/订阅模式

• 主题（Topic）：消息被发送到主题，所有订阅该主题的消费者都会收到消息。
• 特点：消息可被多个消费者同时消费。
• 应用场景：广播消息通知，如新闻推送、股票行情。

2.2 消息持久化机制

为了确保在系统故障时消息不丢失，ActiveMQ支持将消息持久化到存储介质。

2.2.1 KahaDB

• 简介：ActiveMQ默认的文件存储方式，专为消息存储优化。
• 特点：高性能、占用磁盘空间小、恢复速度快。

2.2.2 JDBC持久化

• 简介：将消息持久化到关系型数据库中。
• 特点：利用数据库的可靠性和备份机制，但性能可能较文件存储略低。

2.2.3 LevelDB与其他存储

• LevelDB：高性能的键值存储，适用于对性能要求较高的场景。
• 自定义存储：通过插件机制，可接入自定义的存储方案。

2.3 事务和确认机制

2.3.1 事务支持

• JMS事务：ActiveMQ支持JMS事务，确保一系列消息操作的原子性。
• 使用场景：需要确保多条消息要么全部成功，要么全部失败的业务逻辑。

2.3.2 消息确认模式

• 自动确认（AUTO_ACKNOWLEDGE）：消费者成功接收消息后，自动向Broker发送确认。
• 客户端确认（CLIENT_ACKNOWLEDGE）：由客户端代码控制何时发送确认。
• 事务性会话（SESSION_TRANSACTED）：在事务会话中，确认由commit或rollback控制。

2.4 ActiveMQ的架构组件

• Broker：消息代理服务器，负责接收、路由、存储和转发消息。
• Producer：消息生产者，创建并发送消息到Broker。
• Consumer：消息消费者，从Broker接收并处理消息。
• Destination：消息目的地，Queue或Topic。

三、ActiveMQ的实践案例

为了更好地理解ActiveMQ的实际应用，下面将通过具体案例展示如何使用ActiveMQ进行消息传递。

3.1 环境搭建

3.1.1 安装ActiveMQ

步骤一：下载ActiveMQ

wget https://archive.apache.org/dist/activemq/5.17.1/apache-activemq-5.17.1-bin.tar.gz

步骤二：解压文件

tar -zxvf apache-activemq-5.17.1-bin.tar.gz

步骤三：启动服务

cd apache-activemq-5.17.1
bin/activemq start

3.1.2 验证安装

在浏览器中访问管理控制台：

http://localhost:8161/admin

使用默认用户名和密码admin/admin登录。

3.2 基于Java的消息生产者和消费者

3.2.1 依赖引入

在pom.xml中添加ActiveMQ的依赖：

<dependency>
    <groupId>org.apache.activemq</groupId>
    <artifactId>activemq-client</artifactId>
    <version>5.17.1</version>
</dependency>

3.2.2 消息生产者示例

import javax.jms.*;
import org.apache.activemq.ActiveMQConnectionFactory;

public class Producer {
    public static void main(String[] args) {
        try {
            // 创建连接工厂
            ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
            // 创建连接
            Connection connection = factory.createConnection();
            // 启动连接
            connection.start();
            // 创建会话（不使用事务，自动确认）
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            // 创建队列
            Destination destination = session.createQueue("TestQueue");
            // 创建消息生产者
            MessageProducer producer = session.createProducer(destination);
            // 设置消息持久化
            producer.setDeliveryMode(DeliveryMode.PERSISTENT);
            // 创建消息
            TextMessage message = session.createTextMessage("Hello ActiveMQ");
            // 发送消息
            producer.send(message);
            System.out.println("消息已发送：" + message.getText());
            // 关闭资源
            producer.close();
            session.close();
            connection.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

3.2.3 消息消费者示例

import javax.jms.*;
import org.apache.activemq.ActiveMQConnectionFactory;

public class Consumer {
    public static void main(String[] args) {
        try {
            // 创建连接工厂
            ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
            // 创建连接
            Connection connection = factory.createConnection();
            // 启动连接
            connection.start();
            // 创建会话（不使用事务，自动确认）
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            // 创建队列
            Destination destination = session.createQueue("TestQueue");
            // 创建消息消费者
            MessageConsumer consumer = session.createConsumer(destination);
            // 设置消息监听器
            consumer.setMessageListener(new MessageListener() {
                public void onMessage(Message message) {
                    try {
                        if (message instanceof TextMessage) {
                            TextMessage textMessage = (TextMessage) message;
                            System.out.println("收到消息：" + textMessage.getText());
                        }
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            });
            // 保持程序运行
            System.in.read();
            // 关闭资源
            consumer.close();
            session.close();
            connection.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

代码解析：

  在本次的代码演示中，我将会深入剖析每句代码，详细阐述其背后的设计思想和实现逻辑。通过这样的讲解方式，我希望能够引导同学们逐步构建起对代码的深刻理解。我会先从代码的结构开始，逐步拆解每个模块的功能和作用，并指出关键的代码段，并解释它们是如何协同运行的。通过这样的讲解和实践相结合的方式，我相信每位同学都能够对代码有更深入的理解，并能够早日将其掌握，应用到自己的学习和工作中。

import javax.jms.*;
import org.apache.activemq.ActiveMQConnectionFactory;

public class Consumer {
    public static void main(String[] args) {
        try {
            // 创建连接工厂
            ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
            // 创建连接
            Connection connection = factory.createConnection();
            // 启动连接
            connection.start();
            // 创建会话（不使用事务，自动确认）
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            // 创建队列
            Destination destination = session.createQueue("TestQueue");
            // 创建消息消费者
            MessageConsumer consumer = session.createConsumer(destination);
            // 设置消息监听器
            consumer.setMessageListener(new MessageListener() {
                public void onMessage(Message message) {
                    try {
                        if (message instanceof TextMessage) {
                            TextMessage textMessage = (TextMessage) message;
                            System.out.println("收到消息：" + textMessage.getText());
                        }
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            });
            // 保持程序运行
            System.in.read();
            // 关闭资源
            consumer.close();
            session.close();
            connection.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

这段代码是使用ActiveMQ实现一个简单的消息消费者（Consumer），用于从消息队列中接收和处理消息。下面将对代码进行逐行解析，帮助理解其工作原理。

导入必要的类

import javax.jms.*;
import org.apache.activemq.ActiveMQConnectionFactory;

• javax.jms.*：导入JMS（Java消息服务）API中的所有类和接口。
• org.apache.activemq.ActiveMQConnectionFactory：ActiveMQ提供的连接工厂实现，用于创建连接到ActiveMQ消息代理（Broker）的连接。

定义主类

public class Consumer {
    public static void main(String[] args) {
        // ...
    }
}

• 定义一个名为Consumer的公共类，其中包含main方法，程序的入口点。

创建连接工厂

ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");

• 作用：创建连接到ActiveMQ的连接工厂。
• 参数："tcp://localhost:61616"是ActiveMQ默认的TCP连接URL，表示连接到本地的ActiveMQ服务器。

创建连接并启动

Connection connection = factory.createConnection();
connection.start();

• 创建连接：使用连接工厂创建与Broker的连接。
• 启动连接：必须在创建会话前启动连接，以便开始接收消息。

创建会话

Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

• 参数解释：
  • false：表示不使用事务。
  • Session.AUTO_ACKNOWLEDGE：消息自动确认，消费者成功接收消息后，自动向Broker发送确认。

创建目的地（队列）

Destination destination = session.createQueue("TestQueue");

• 作用：指定要消费的队列名称为"TestQueue"。
• 说明：如果队列不存在，ActiveMQ会自动创建。

创建消息消费者

MessageConsumer consumer = session.createConsumer(destination);

• 作用：创建一个消费者，用于从指定的目的地接收消息。

设置消息监听器

consumer.setMessageListener(new MessageListener() {
    public void onMessage(Message message) {
        try {
            if (message instanceof TextMessage) {
                TextMessage textMessage = (TextMessage) message;
                System.out.println("收到消息：" + textMessage.getText());
            }
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
});

• 作用：为消费者设置一个消息监听器，当有消息到达时自动调用onMessage方法。
• 逻辑解析：
  • 类型检查：判断消息是否为TextMessage类型。
  • 消息处理：如果是文本消息，获取消息内容并打印出来。
  • 异常处理：捕获并打印JMSException。

保持程序运行

System.in.read();

• 作用：程序阻塞，等待用户输入。这样可以使程序持续运行，保持对消息的监听。
• 说明：在没有其他操作时，程序会等待用户按下任意键后继续执行后续代码。

关闭资源

consumer.close();
session.close();
connection.close();

• 作用：依次关闭消息消费者、会话和连接，释放资源。
• 位置：这些代码在用户按下任意键后执行，表示程序准备结束。

异常处理

} catch (Exception e) {
    e.printStackTrace();
}

• 作用：捕获整个try块中可能抛出的任何异常，打印堆栈信息，方便调试。

总体流程

1. 初始化连接和会话：创建与ActiveMQ的连接，启动连接，创建会话。
2. 指定目的地：创建或指定要消费的队列TestQueue。
3. 创建消费者并设置监听器：创建消费者，并通过setMessageListener方法设置消息监听器，实现异步接收消息。
4. 处理消息：当有新消息到达时，onMessage方法被调用，程序处理消息并输出内容。
5. 保持监听状态：通过System.in.read()方法使程序阻塞，持续监听消息。
6. 关闭资源：当程序结束时，关闭消费者、会话和连接，释放资源。

注意事项

• 线程安全：JMS会话不是线程安全的，建议每个线程使用独立的会话。
• 消息类型：在处理消息时，需要根据实际情况判断消息类型，例如TextMessage、BytesMessage等。
• 异常处理：应完善异常处理机制，避免程序因未处理的异常而崩溃。
• 资源释放：确保在程序结束前正确关闭所有JMS资源。

执行效果

• 当有生产者向TestQueue队列发送消息后，该消费者会自动接收到消息并打印。

• 例如，生产者发送了消息"Hello ActiveMQ"，消费者将输出：

  收到消息：Hello ActiveMQ
  

扩展思考

• 多线程消费：可以创建多个消费者实例，利用多线程提高消息消费的吞吐量。
• 消息过滤：在创建消费者时，添加消息选择器（Message Selector）以过滤特定的消息。
• 事务支持：将会话设置为事务性会话，以确保消息处理的原子性和可靠性。
• 手动确认：将确认模式设置为CLIENT_ACKNOWLEDGE，由客户端代码手动确认消息，增强对消息消费的控制。

示例改进

使用Lambda表达式（Java 8及以上）

如果使用Java 8及以上版本，可以使用Lambda表达式简化代码：

consumer.setMessageListener(message -> {
    try {
        if (message instanceof TextMessage) {
            TextMessage textMessage = (TextMessage) message;
            System.out.println("收到消息：" + textMessage.getText());
        }
    } catch (JMSException e) {
        e.printStackTrace();
    }
});

手动确认消息

// 创建会话（不使用事务，手动确认）
Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);

// 在消息处理后手动确认
textMessage.acknowledge();

添加异常重试机制

在onMessage方法中，添加对消息处理失败的重试机制，增强程序的健壮性。

3.3 高级特性应用

3.3.1 消息过滤

通过消息选择器（Message Selector）实现基于属性的消息过滤。

生产者发送消息并设置属性：

TextMessage message = session.createTextMessage("User Info");
message.setIntProperty("age", 30);
producer.send(message);

消费者基于选择器接收消息：

MessageConsumer consumer = session.createConsumer(destination, "age >= 18");

3.3.2 事务处理

在需要保证消息操作原子性的情况下，使用事务会话。

创建事务会话：

Session session = connection.createSession(true, Session.SESSION_TRANSACTED);

提交或回滚事务：

// 提交事务
session.commit();

// 或者回滚事务
session.rollback();

3.3.3 延迟与定时消息

ActiveMQ支持消息的延迟投递和定时发送。

设置延迟投递：

message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, 5000);
producer.send(message);

3.3.4 持久订阅

在发布/订阅模式下，使用持久订阅可以在消费者不在线时保留消息。

创建持久订阅者：

TopicSubscriber subscriber = session.createDurableSubscriber(topic, "SubscriberName");

3.4 ActiveMQ的监控与管理

3.4.1 使用JMX监控

ActiveMQ支持通过JMX（Java Management Extensions）进行监控。

• 启用JMX：默认情况下已启用，可通过jconsole连接。
• 监控内容：队列大小、消费者数量、消息堆积等。

3.4.2 Web控制台

通过Web控制台，可以直观地查看Broker的运行状态。

• 查看队列和主题：消息数量、消费者、生产者等信息。
• 管理操作：创建或删除队列、清除消息等。

四、ActiveMQ的拓展应用

4.1 集群与高可用配置

4.1.1 主从模式

通过配置主从Broker，实现故障转移和高可用性。

• 共享文件系统：主从Broker共享同一套KahaDB文件。
• ZooKeeper：使用ZooKeeper协调主从Broker的角色。

4.1.2 网络连接器

使用网络连接器将多个Broker连接成网络拓扑，提供负载均衡和故障转移。

<networkConnectors>
    <networkConnector uri="static:(tcp://remotehost:61616)" />
</networkConnectors>

4.2 与Spring集成

4.2.1 配置Spring JMS

在Spring应用中，使用JmsTemplate简化消息的发送和接收。

配置连接工厂：

<bean id="connectionFactory" class="org.apache.activemq.spring.ActiveMQConnectionFactory">
    <property name="brokerURL" value="tcp://localhost:61616" />
</bean>

配置JmsTemplate：

<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
    <property name="connectionFactory" ref="connectionFactory" />
</bean>

4.2.2 使用消息监听器容器

<jms:listener-container connection-factory="connectionFactory">
    <jms:listener destination="TestQueue" ref="messageListener" />
</jms:listener-container>

4.3 安全与认证

4.3.1 基于JAAS的认证

通过配置login.config文件，使用JAAS进行用户认证。

配置示例：

activemq {
    org.apache.activemq.jaas.PropertiesLoginModule required
    org.apache.activemq.jaas.properties.user="users.properties"
    org.apache.activemq.jaas.properties.group="groups.properties";
};

4.3.2 权限控制

在activemq.xml中配置授权插件，控制用户对队列和主题的访问权限。

<plugins>
    <authorizationPlugin>
        <map>
            <authorizationMap>
                <authorizationEntries>
                    <authorizationEntry queue=">" read="users" write="users" admin="admins"/>
                </authorizationEntries>
            </authorizationMap>
        </map>
    </authorizationPlugin>
</plugins>

4.4 性能调优

4.4.1 系统资源优化

• JVM参数：根据系统资源，合理设置堆内存大小。
• 磁盘IO：使用SSD或优化磁盘配置，提高存储性能。
• 网络配置：确保网络带宽和稳定性，减少延迟。

4.4.2 Broker参数调整

• 线程池大小：调整maxConnections和maximumActive参数。
• 消息预取：根据消费者处理能力，调整prefetch大小。
• 持久化策略：选择合适的持久化机制，平衡性能与可靠性。

4.5 常见问题与故障排除

4.5.1 消息堆积

• 原因分析：消费者处理能力不足，消息消费速度慢。
• 解决方案：增加消费者数量，优化消费者处理逻辑。

4.5.2 连接异常

• 原因分析：网络不稳定、Broker负载过高。
• 解决方案：检查网络状况，优化Broker配置，使用连接池。

4.5.3 消息丢失

• 原因分析：未正确配置消息持久化或事务。
• 解决方案：确保消息持久化配置正确，使用事务会话。

五、ActiveMQ与其他消息中间件的比较

5.1 与RabbitMQ的比较

• 协议支持：RabbitMQ主要基于AMQP协议，而ActiveMQ支持多种协议。
• 性能：RabbitMQ在高并发短消息处理上表现优异，ActiveMQ在大消息传输上有优势。
• 生态系统：ActiveMQ在Java领域有更好的集成支持，RabbitMQ在Erlang社区中更活跃。

5.2 与Kafka的比较

• 定位不同：Kafka定位为分布式流平台，适合大数据实时处理。
• 消息模型：Kafka不遵循JMS规范，采用发布/订阅模式，不支持队列模式。
• 性能：Kafka在高吞吐量、持久化日志方面性能卓越。

六、实际案例分析

6.1 电子商务平台的应用

在某大型电子商务平台中，ActiveMQ被用于以下场景：

• 订单处理：订单创建后，异步通知库存、物流等系统。
• 消息通知：向用户发送下单成功、发货等短信或邮件通知。
• 日志收集：收集用户行为日志，供大数据分析使用。

效果：

• 解耦系统：各模块独立开发、部署，降低了系统复杂度。
• 提升性能：通过异步处理，减少了请求的响应时间。
• 提高可靠性：消息持久化和事务支持，确保关键数据不丢失。

6.2 金融行业的应用

在某银行的分布式系统中，ActiveMQ用于：

• 交易处理：实时传递交易指令，确保资金结算的及时性。
• 风险控制：异步收集风险数据，实时监控系统风险。
• 数据同步：跨地域的数据中心之间的数据同步。

效果：

• 降低延迟：高性能的消息传递，满足金融业务的实时性要求。
• 提高安全性：通过安全认证和权限控制，保障数据安全。
• 增强扩展性：灵活的集群部署，满足业务增长的需求。

七、总结

ActiveMQ作为一款成熟、稳定、功能丰富的开源消息中间件，在企业级应用中发挥着重要作用。通过深入理解ActiveMQ的核心原理和实践应用，开发者可以有效地解决系统解耦、异步通信、流量削峰等问题。同时，结合实际业务场景，合理利用ActiveMQ的高级特性，如事务、持久化、集群部署等，可以大幅提升系统的性能和可靠性。

在未来的系统架构设计中，消息中间件将继续扮演关键角色。希望本文能帮助读者深入理解ActiveMQ，灵活应用于实际项目中，加速业务的发展与创新。

参考资料

• Apache ActiveMQ 官方文档
• Java Message Service (JMS) 规范
• ActiveMQ In Action
• Spring JMS 官方文档
• 消息队列之道：深入理解与实践

☀️建议/推荐你

无论你是计算机专业的学生，还是对编程有兴趣的小伙伴，都建议直接毫无顾忌的学习此专栏「滚雪球学Java」，bug菌郑重承诺，凡是学习此专栏的同学，均能获取到所需的知识和技能，全网最快速入门Java编程，就像滚雪球一样，越滚越大，指数级提升。

码字不易，如果这篇文章对你有所帮助，帮忙给bug菌来个一键三连(关注、点赞、收藏) ，您的支持就是我坚持写作分享知识点传播技术的最大动力。
  同时也推荐大家关注我的硬核公众号:「猿圈奇妙屋」 ；以第一手学习bug菌的首发干货，不仅能学习更多技术硬货，还可白嫖最新BAT大厂面试真题、4000G Pdf技术书籍、万份简历/PPT模板、技术文章Markdown文档等海量资料，你想要的我都有！

📣关于我

我是bug菌，CSDN | 掘金 | infoQ | 51CTO 等社区博客专家，历届博客之星Top30，掘金年度人气作者Top40，51CTO年度博主Top12，掘金等平台签约作者，华为云 | 阿里云| 腾讯云等社区优质创作者，全网粉丝合计30w+ ；硬核微信公众号「猿圈奇妙屋」，欢迎你的加入！免费白嫖最新BAT互联网公司面试题、4000G pdf电子书籍、简历模板等海量资料。

–End