ActiveMQ 基础实践指南
ActiveMQ 是由 Apache 开发的一款开源消息中间件，完全遵循 JMS 1.1 和 J2EE 1.4 规范，为分布式系统提供异步通信能力。作为消息队列的成熟实现，它在企业级应用中仍扮演关键角色。
JMS（Java Message Service）是 Java 平台上用于面向消息中间件的 API，支持跨应用和分布式系统的消息传递。与 JDBC 类似，JMS 提供厂商无关的接口，可对接 IBM MQSeries、BEA Weblogic JMS 等不同实现。
本文演示 ActiveMQ 的基本使用。首先从 官网 下载并解压，确保已配置 Java 环境。启动服务后，访问
http://localhost:8161/admin，使用默认凭据 admin/admin 登录（凭据在 conf/users.properties 中定义）。登录界面如下：
接下来，通过 Maven 项目引入依赖：
<dependencies>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-all</artifactId>
<version>${active-mq-version}</version>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.12</version>
<scope>test</scope>
</dependency>
</dependencies>点对点消息收发
1. 消息发送
实现步骤：创建连接工厂 → 建立连接 → 开启会话 → 定义队列 → 生成生产者 → 发送消息。
首先初始化组件：
private static ConnectionFactory factory;
private static Connection connection;
private static Session session;
private static Destination destination;
private static MessageProducer producer;创建连接工厂并配置默认参数：
private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
private static final String BROKER_URL = ActiveMQConnection.DEFAULT_BROKER_URL;
factory = new ActiveMQConnectionFactory(USERNAME, PASSWORD, BROKER_URL);发送方法实现：
public static boolean sendMessage(String queueName, String content) {
boolean success = true;
try {
connection = factory.createConnection();
connection.start();
session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
destination = session.createQueue(queueName);
producer = session.createProducer(destination);
TextMessage msg = session.createTextMessage(content);
producer.send(msg);
session.commit();
} catch (JMSException e) {
e.printStackTrace();
success = false;
} finally {
if (connection != null) {
try { connection.close(); } catch (JMSException e) { e.printStackTrace(); }
}
}
return success;
}createSession 参数说明：
- 第一个参数（事务）：
true启用事务，忽略第二个参数；false禁用事务。 - 第二个参数（确认模式）：
Session.AUTO_ACKNOWLEDGE：自动确认Session.CLIENT_ACKNOWLEDGE：客户端调用acknowledge()确认Session.DUPS_OK_ACKNOWLEDGE：允许重复确认
2. 消息接收
步骤与发送类似，但使用消费者替代生产者。
public static void receiveMessage(String queueName) {
try {
connection = factory.createConnection();
connection.start();
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
destination = session.createQueue(queueName);
MessageConsumer consumer = session.createConsumer(destination);
while (true) {
TextMessage msg = (TextMessage) consumer.receive();
if (msg != null) {
System.out.println(queueName + " 收到消息：" + msg.getText());
}
}
} catch (Exception e) {
System.err.println("消息接收异常");
} finally {
if (connection != null) {
try { connection.close(); } catch (JMSException e) { e.printStackTrace(); }
}
}
}receive() 提供了三种阻塞模式：receive()（持续等待）、receive(1000)（等待1秒）、receiveNoWait()（不等待）。
3. 使用监听器
通过 MessageListener 接口实现异步监听：
public class MsgListener implements MessageListener {
private String queueName;
public MsgListener(String queueName) {
this.queueName = queueName;
}
@Override
public void onMessage(Message msg) {
if (msg instanceof TextMessage) {
try {
System.out.println("【" + queueName + "】消息：" + ((TextMessage) msg).getText());
} catch (JMSException e) {
System.err.println("消息解析失败");
}
}
}
}改造接收方法，应用监听器：
public static void receiveWithListener(String queueName) {
try {
connection = factory.createConnection();
connection.start();
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
destination = session.createQueue(queueName);
MessageConsumer consumer = session.createConsumer(destination);
consumer.setMessageListener(new MsgListener(queueName));
Thread.sleep(Long.MAX_VALUE); // 保持监听
} catch (Exception e) {
System.err.println("监听异常");
} finally {
if (connection != null) {
try { connection.close(); } catch (JMSException e) { e.printStackTrace(); }
}
}
}发布订阅模式
相比点对点模式，唯一区别是使用 session.createTopic() 替代 session.createQueue()：
destination = session.createTopic(topicName);其余逻辑（连接创建、会话配置、生产者/消费者使用）保持一致，此处不再重复。完整示例代码可参考 GitHub 仓库。
