ActiveMQ是一个消息队列应用服务器（推送服务器）。支持JMS规范。

JMS 全称：Java Message Service ，即为 Java 消息服务，是一套 java 消息服务的 API 接口。实现了 JMS 标准的系统，称之为 JMS Provider。

消息队列是在消息的传输过程中保存消息的容器，提供一种不同进程或者同一进程不同线程直接通讯的方式。

常见消息队列应用：

ActiveMQ 是Apache出品，最流行的，能力强劲的开源消息总线。ActiveMQ 是一个完全支持JMS1.1和J2EE 1.4规范的 JMS Provider实现。

RabbitMQ是一个在AMQP基础上完成的，可复用的企业消息系统。他遵循Mozilla Public License开源协议。开发语言为Erlang。

由阿里巴巴定义开发的一套消息队列应用服务。

使用 ActiveMQ 实现消息队列模型。

默认端口为: 8161

默认用户名密码均为 admin

注意：若防火墙没有配置该服务的端口，必须在防火墙中配置。

点击图示选项后，进行登录

然后成功进入管理界面

ActiveMQ 的解压包中，提供了运行 ActiveMQ 需要的 jar 包。

ActiveMQ 是实现了 JMS 规范的。在实现消息服务的时候，必须基于 API 接口规范。

下述 API 都是接口类型，定义在 javax.jms 包中，是 JMS 标准接口定义。ActiveMQ 完全实现这一套 api 标准。

链接工厂, 用于创建链接的工厂类型。

链接，用于建立访问ActiveMQ连接的类型, 由链接工厂创建。

会话, 一次持久有效、有状态的访问，由链接创建。

目的地, 即本次访问ActiveMQ消息队列的地址，由Session会话创建。

消息，在消息传递过程中数据载体对象，是所有消息【文本消息TextMessage，对象消息ObjectMessage等】具体类型的顶级接口，可以通过会话创建或通过会话从 ActiveMQ 服务中获取。

消息生成者, 在一次有效会话中, 用于发送消息给ActiveMQ服务的工具，由Session会话创建。

消息消费者【消息订阅者，消息处理者】, 在一次有效会话中, 用于ActiveMQ服务中获取消息的工具，由Session会话创建。

注意：ActiveMQ 服务接受消息的入口是 61616 端口，防火墙还需要开放此端口

import org.apache.activemq.ActiveMQConnectionFactory;import org.junit.Test;

import javax.jms.*;

/** * @version 1.0 * @program: spring * @description: 消息生产者 * @author: chris * @create: 2022-04-21 10:30 * @since JDK1.8 **/public class MyProducer {

//定义连接工厂 ConnectionFactory connectionFactory;

//定义连接 Connection connection;

//定义session会话 Session session;

//定义消息目的地 Destination destination;

//定义消息生产者 MessageProducer producer;

//定义消息 Message message;

public void init() throws JMSException { //创建连接工厂 connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616"); //创建连接 connection = connectionFactory.createConnection(); //启动连接 connection.start(); //创建session会话，第一个参数是是否开启事务，第二个参数是自动确认模式 session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); //创建消息目的地，Queue：消息类型是队列，Topic：消息类型是主题 destination = session.createQueue("myQueue"); //创建消息生产者 producer = session.createProducer(destination); //设置消息生产者的持久化 // producer.setDeliveryMode(DeliveryMode.PERSISTENT); //创建消息 message = session.createTextMessage("hello world"); }

public void sendMessage(String message) throws JMSException { //创建消息 TextMessage textMessage = session.createTextMessage(message); //发送消息 producer.send(textMessage); }

public void close() throws JMSException { //关闭消息生产者 producer.close(); //关闭会话 session.close(); //关闭连接 connection.close(); }

@Test public void sentToActiveMQ(){ try { /* * 创建连接工厂，由 ActiveMQ 实现。构造方法参数 * userName 用户名 * password 密码 * brokerURL 访问 ActiveMQ 服务的路径地址，结构为: 协议名://主机地址:端口号 */ connectionFactory = new ActiveMQConnectionFactory("admin", "admin", "tcp://127.0.0.1:61616"); //创建连接 connection = connectionFactory.createConnection(); //启动连接 connection.start(); /* * 创建会话，参数含义: * 1.transacted - 是否使用事务 * 2.acknowledgeMode - 消息确认机制，可选机制为： * 1）Session.AUTO_ACKNOWLEDGE - 自动确认消息 * 2）Session.CLIENT_ACKNOWLEDGE - 客户端确认消息机制 * 3）Session.DUPS_OK_ACKNOWLEDGE - 有副本的客户端确认消息机制 */ session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); //创建消息目的地，也就是队列名，Queue：消息类型是队列，Topic：消息类型是主题 destination = session.createQueue("myQueue"); //创建消息生产者 producer = session.createProducer(destination); //创建消息 message = session.createTextMessage("hello world"); //发送消息 producer.send(message); } catch (JMSException e) { e.printStackTrace(); } finally { try { //关闭消息生产者 if (producer != null) { producer.close(); } //关闭会话 if (session != null) { session.close(); } //关闭连接 if (connection != null) { connection.close(); } } catch (JMSException e) { e.printStackTrace(); } } }}

使用测试方法执行后，可以看到成功将消息加入队列

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

/** * @version 1.0 * @program: spring * @description: 消费者 * @author: chris * @create: 2022-04-21 10:30 * @since JDK1.8 **/public class MyConsumer {

//定义连接工厂 ConnectionFactory connectionFactory;

//定义连接 Connection connection;

//定义会话 Session session;

//定义目的地 Destination destination;

//定义消费者 MessageConsumer consumer;

//定义消息 Message message;

public void init() throws Exception { //创建连接工厂 connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616"); //创建连接 connection = connectionFactory.createConnection(); //启动连接 connection.start(); //创建会话 session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); //创建目的地 destination = session.createQueue("myQueue"); //创建消费者 consumer = session.createConsumer(destination);

//接收消息 message = consumer.receive(); }

public void close() throws Exception { //关闭消费者 consumer.close(); //关闭会话 session.close(); //关闭连接 connection.close(); }

public static void main(String[] args) throws Exception { MyConsumer myConsumer = new MyConsumer(); myConsumer.init(); System.out.println("接收到的消息是：" + ((TextMessage) myConsumer.message).getText()); myConsumer.close(); }}

测试结果

且后台管理界面也可以看到已被消费

在前面的示例中，我们发现消费者每次只能消费一条消息。当队列中有多条消息的时候，我们需要多次运行消费者，才能消费完这些消息。我们希望一次将所有的消息全部接收，可以使用 ActiveMQ 监听器来监听队列，持续消费消息。

答：使用ActiveMQ监听器来监听队列，持续消费消息。

自定义监听器需要实现 MessageListener 接口

import javax.jms.Message;import javax.jms.MessageListener;import javax.jms.TextMessage;

/** * @version 1.0 * @program: spring * @description: 消息监听器 * @author: chris * @create: 2022-04-21 10:30 * @since JDK1.8 **/public class MyListener implements MessageListener {

@Override public void onMessage(Message message) { try { TextMessage textMessage = (TextMessage) message; System.out.println("收到消息：" + textMessage.getText()); } catch (Exception e) { e.printStackTrace(); } }

}

监听器需要持续加载，因此需要使消费程序不结束。这里我们使用输入流阻塞消费线程结束。（实际开发中，使用web项目加载）

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

/** * @version 1.0 * @program: spring * @description: 消费者 * @author: chris * @create: 2022-04-21 10:30 * @since JDK1.8 **/public class MyConsumer {

//定义连接工厂 ConnectionFactory connectionFactory;

//定义连接 Connection connection;

//定义会话 Session session;

//定义目的地 Destination destination;

//定义消费者 MessageConsumer consumer;

//定义消息 Message message;

public void init() throws Exception { //创建连接工厂 connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616"); //创建连接 connection = connectionFactory.createConnection(); //启动连接 connection.start(); //创建会话 session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); //创建目的地 destination = session.createQueue("myQueue"); //创建消费者 consumer = session.createConsumer(destination);

//接收消息 // message = consumer.receive();

//加载监听器 consumer.setMessageListener(new MyListener());

//监听器需要持续加载，所以需要一直调用，使用输入流的方式阻塞当前线程结束 System.in.read(); }

public void close() throws Exception { //关闭消费者 consumer.close(); //关闭会话 session.close(); //关闭连接 connection.close(); }

public static void main(String[] args) throws Exception { MyConsumer myConsumer = new MyConsumer(); myConsumer.init(); System.out.println("接收到的消息是：" + ((TextMessage) myConsumer.message).getText()); myConsumer.close(); }}

先添加 5 条消息到队列

运行 Consumer 的测试程序，可以看到连续接受了 5 条消息，且接续添加会继续输出消息。

在入门示例中，只能向一个消费者发送消息。但是有一些场景，需求有多个消费者都能接收到消息，比如：美团 APP 每天的消息推送。该如何实现呢？

ActiveMQ是通过不同的服务模式来解决这个问题的。要搞清楚这个问题，必须知道ActiveMQ有哪些应用模式。

消息模型

入门示例就是采用的这种 PTP 服务模式

消息模型

所以，主题订阅模式下，==一定要先有消息的消费者(订阅者)，后有消息的生产者(发布者)==。

主题订阅模式下，可以有多个订阅者。我们这里用多线程来模拟。创建 MySubscriber 类，实现 Runnable 接口

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

/** * @version 1.0 * @program: spring * @description: 主题订阅者 * @author: chris * @create: 2022-04-21 10:52 * @since JDK1.8 **/public class MySubscriber implements Runnable {

// 定义连接工厂 TopicConnectionFactory factory;

// JMS连接，属于Pub/Sub方式的连接 TopicConnection connection;

// JMS会话，属于Pub/Sub方式的会话 TopicSession session;

// 定义消息队列，消息队列名称为：TEST_QUEUE Topic topic;

// 主题订阅者 TopicSubscriber subscriber;

// 定义消息 Message message;

@Override public void run() { try { // 创建连接工厂 factory = new ActiveMQConnectionFactory("tcp://localhost:61616"); // 通过工厂创建连接 connection = factory.createTopicConnection(); // 开启连接 connection.start(); // 创建会话 session = connection.createTopicSession(false, TopicSession.AUTO_ACKNOWLEDGE); // 创建消息队列 参数：主题名称，是否独占，是否支持事务（topic模型） topic = session.createTopic("topic"); // 创建订阅者 subscriber = session.createSubscriber(topic); // 接收消息，会阻塞线程 message = subscriber.receive(); // 打印消息 System.out.println("收到消息：" + ((TextMessage) message).getText()); } catch (Exception e) { e.printStackTrace(); } finally { try { // 关闭资源 if (subscriber != null) { subscriber.close(); } if (session != null) { session.close(); } if (connection != null) { connection.close(); } } catch (JMSException e) { e.printStackTrace(); } } }}

注意 ==junit 不支持多线程测试==，需要使用 main 方法执行测试

/** * @version 1.0 * @program: spring * @description: Topic模式测试类 * @author: chris * @create: 2022-04-21 11:03 * @since JDK1.8 **/public class ActiveMQTest {

public static void main(String[] args) { MySubscriber subscriber = new MySubscriber(); Thread t1 = new Thread(subscriber); Thread t2 = new Thread(subscriber); t1.start(); t2.start(); }}

启动程序后可以从管理面板看到该 topic 下有两个订阅者

创建 MyPublish 类

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

/** * @version 1.0 * @program: spring * @description: 主题发布者 * @author: chris * @create: 2022-04-21 11:07 * @since JDK1.8 **/public class MyPublisher {

// 定义连接工厂 TopicConnectionFactory factory;

// 定义连接 TopicConnection connection;

// 定义会话 TopicSession session;

// 定义发布者 TopicPublisher publisher;

// 定义主题 Topic topic;

// 定义消息 Message message;

public void publishTopic(String topicName, String publishText) { try { // 创建连接工厂 factory = new ActiveMQConnectionFactory("tcp://localhost:61616"); // 通过连接工厂创建连接 connection = factory.createTopicConnection(); // 启动连接 connection.start(); // 创建会话 session = connection.createTopicSession(false, TopicSession.AUTO_ACKNOWLEDGE); // 创建主题 topic = session.createTopic(topicName); // 创建发布者 publisher = session.createPublisher(topic); // 创建消息 message = session.createTextMessage(publishText); // 发送消息 publisher.send(message); } catch (Exception e) { e.printStackTrace(); } finally { try { // 关闭资源 if (null != publisher) { publisher.close(); } if (null != session) { session.close(); } if (null != connection) { connection.close(); } } catch (JMSException e) { e.printStackTrace(); } } }}

新增测试方法，发布主题消息

import org.junit.Test;

/** * @version 1.0 * @program: spring * @description: 发布Topic消息测试类 * @author: chris * @create: 2022-04-21 11:18 * @since JDK1.8 **/public class PublisherTest {

@Test public void topicPublishTest() { MyPublisher publisher = new MyPublisher(); publisher.publishTopic("topic", "hello, topic"); }}

执行 topicPublisherTest，可以看到订阅者立即收到了消息

同时在管理页面也能看到一条消息进入，被读取了两次。

当队列中有未被消费的消息时，我们重新启动ActiveMQ服务器后，发现消息仍然在队列中。

ActiveMQ 是支持持久化的，可以永久保存消息。消息是保存在内存中的。当内存空间不足，或者ActiveMQ服务关闭的时候，消息会被持久化到磁盘上。被消费的时候，再加载到内存空间中。

ActiveMQ持久化方式在 activemq/conf/activemq.xml 中指定

kahadb 方式是 ActiveMQ 默认的持久化策略。不会保存已经被消费过的消息。从配置文件中可以看到默认的存储地址，也就是 /usr/local/activemq/data/kahadb

5.3 版本之前，现在已经过时，不考虑。

ActiveMQ 可以将数据持久化到数据库中，支持使用任意的数据库。

数据库最好不要跟 ActiveMQ 服务器在同一台机器。因为当 cpu 线程资源不足时，往队列中写入消息时，如果数据库上一次持久化还没结束，容易造成线程阻塞。

这里数据库建立在宿主机上，ActiveMQ 服务部署在虚拟机。

配置数据源时，是支持连接池的，这里使用 druid 作为连接池。将 jdbc 驱动、druid 的 jar 包上传到 activemq/lib/ 目录下

注意在 节点外，创建数据源。

在 节点内部，注释 kahadb 方式，添加 jdbc 方式。

进入 activemq/bin 目录，使用命令 ./activemq restart 重新启动 ActiveMQ。可以看到，数据库中新增了三张表。

运行入门示例中的测试类，往队列中写入一条消息，可以在表activemq_msgs中，看到新写入了一条数据

jdbc 持久化方式，只要Mysql数据库稳定运行，就能保证队列中消息的安全。安全级别高，但是效率低。因此，在实际开发中，除非是像银行这类对数据安全极高的业务，我们一般都是使用默认持久化方式 kahadb。

我们判断一个程序的优劣，有一个很重要的指标：高内聚、低耦合。

但是有一些业务功能，必须涉及到两个不同的业务，那我们就要想办法，尽量将它们解耦。以我们前面学习的 solr 为例，我们知道 solr 的数据来自数据库。这就意味着，当数据库中的商品发生变化时，我们需要同步更新索引库。这个时候我们就可以使用消息队列模型来解耦，添加添加业务和同步索引库业务。

订单处理，就可以由前端应用将订单信息放到队列，后端应用从队列里依次获得消息处理，高峰时的大量订单可以积压在队列里慢慢处理掉。由于同步通常意味着阻塞，而大量线程的阻塞会降低计算机的性能。

日志处理是指将消息队列用在日志处理中，比如Kafka的应用，解决大量日志传输的问题。架构简化如下：

需要：当我们在网站注册的时候，有时候需要认证邮箱或者手机号，这个时候保存数据到数据库之前，需要先等待认证结束。如果说认证程序耗时比较大，会影响影响用户注册的业务。这个时候，我们可以使用消息队列模型，将同步执行的业务，通过队列，变成异步处理。

Spring已经整合了jms规范了(spring-jms.jar)，而ActiveMQ是实现了jms规范的。这就意味着Spring整合ActiveMQ是非常方便的。并且Spring-jms，提供了一个JmsTemplate类，用来简化消息读写的业务代码。Spring整合ActivMQ之后，就可以使用该类，简化开发。

使用Spring整合ActiveMQ，模拟限时抢购下的流量削峰问题。

订单页面：order.jsp

结果页面：success.jsp

创建 Order 类

创建 OrderController 类

访问 order 页面下订单，确定 springmvc 正常工作，然后进行下一步。

在这里，我们注入 JmsTemplate 类，来简化代码。另外要注意：

这里使用监听器模式

修改 springmvc 的配置文件，添加整合 ActiveMQ 配置

重新启动项目，提交多个订单，可以看到控制台持续输出，测试成功。