分布式消息中间件，ActiveMQ 消息中间件进阶与应用详解

小鱼儿 2023-06-15 09:51 193阅读 0赞

**目录**

概述

ActiveMQ进阶与应用详解

一、常见的消息中间件

二、ActiveMQ 是什么 ？

三、ActiveMQ 特性

四、ActiveMQ 的消息传输模式

五、Linux 下 ActiveMQ-5.x 的安装

六、ActiveMQ 的使用

--------------------

> ### 概述 ###

在进行分布式系统构建的过程中，需要进行各应用之间的解耦合、异步通信、流量削峰等，遂通过消息中间件来实现，实现以高效、可靠的消息传递机制来进行与应用无关的数据交互，经过衡量选择了ActiveMQ，本篇文章主要对ActiveMQ进行浅析，通过本篇文章可以知其一二，并能轻松上手应用与实战。

> ### ActiveMQ进阶与应用详解 ###

### 一、常见的消息中间件 ###

市面上常见的消息中间件有：ActiveMQ、RabbitMQ、kafka。

### 二、ActiveMQ 是什么 ？ ###

ActiveMQ是Apache软件基金会所研发的开放源代码的消息中间件，ActiveMQ是一个纯Java编写的应用程序，所以在使用的过程中，需要环境支持Java虚拟机即可执行ActiveMQ，是能力强劲的开源消息总线。ActiveMQ是一个完全支持JMS和J2EE规范的JMS Provider实现，尽管JMS规范出台已经是很久的事情了，但是JMS在当今的J2EE应用中间仍然扮演着特殊的地位。

###    三、ActiveMQ 特性 ###

1、支持多种语言和协议编写客户端。支持的语言包括：Java、C、C++、C\#、Ruby、Perl、Python、PHP等。支持的应用协议包括：OpenWire、Stomp REST、WS Notification、XMPP、AMQP；  
2、完全支持JMS和J2EE 规范 (持久化、XA消息、事务机制)；  
3、对Spring的支持，ActiveMQ可以很容易内嵌到使用Spring的系统里面去，而且也支持Spring相关的特性；  
4、通过了常见J2EE服务器(如 Geronimo、JBoss、GlassFish、WebLogic)的测试，其中通过JCA resource adaptors的配置，可以让ActiveMQ可以自动的部署到任何兼容J2EE商业服务器上；  
5、支持多种传输协议，包括：in-VM、TCP、SSL、NIO、UDP、JGroups、JXTA；  
6、支持通过JDBC和Journal提供高速的消息持久化；  
7、从设计上保证了高性能的集群，客户端-服务器，点对点；  
8、支持Ajax；  
9、支持与Axis的整合；  
10、可以很容易得调用内嵌的JMS provider进行测试；

### 四、ActiveMQ 的消息传输模式 ###

**两种模式：**  
**点对点传输模式：**即一个生产者对应一个消费者；  
**发布-订阅模式：**即一个生产者产生消息并进行发出后，可以由多个已订阅该消息的消费者进行接收。

JMS定义了五种不同的消息正文格式，以及调用的消息类型，允许发送并接收以一些不同形式的数据，提供现有消息格式的一些级别的兼容性：

<table> 
 <tbody> 
  <tr> 
   <td>StreamMessage</td> 
   <td>Java原始值的数据流</td> 
  </tr> 
  <tr> 
   <td>MapMessage</td> 
   <td>一套名称-值对</td> 
  </tr> 
  <tr> 
   <td>TextMessage</td> 
   <td>一个字符串对象</td> 
  </tr> 
  <tr> 
   <td>ObjectMessage</td> 
   <td>一个序列化的Java对象</td> 
  </tr> 
  <tr> 
   <td>BytesMessage</td> 
   <td>一个字节的数据流</td> 
  </tr> 
 </tbody> 
</table>

### 五、Linux 下 ActiveMQ-5.x 的安装 ###

1）、Java 环境配置

由于 ActiveMQ 是由 Java 语言进行开发的，所以 ActiveMQ 的运行需要依赖于 JDK 环境。

参考【[Linux下Jdk1.8的安装与环境变量配置教程][Linux_Jdk1.8]】

2）、下载 ActiveMQ

官网下载：【[activemq.apache][]】

3）、安装 ActiveMQ

将下载好的 ActiveMQ 程序包上传到 Linux 服务器并解压，如下图：

![20191202104629499.png][]

4）、ActiveMQ 的启动/关闭/状态查看

启动服务：

[root@localhost bin]# ./activemq start

状态查看：

[root@localhost bin]# ./activemq status

关闭服务

[root@localhost bin]# ./activemq stop

如下图：

![20191204110555574.png][]

5）、ActiveMQ 管理界面

访问地址：[http://192.168.159.129:8161/index.html][http_192.168.159.129_8161_index.html]

ActiveMQ 管理界面入口，点击 “ Manage ActiveMQ broker ” ，如下图：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0hlbGxvX1dvcmxkX1FXUA_size_16_color_FFFFFF_t_70][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0hlbGxvX1dvcmxkX1FXUA_size_16_color_FFFFFF_t_70 1][]

初始用户名：admin

初始密码：admin

管理界面如下图：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0hlbGxvX1dvcmxkX1FXUA_size_16_color_FFFFFF_t_70 2][]

### 六、ActiveMQ 的使用 ###

1）、队列（Queue）的使用

Producer 创建步骤（负责生产消息）：

1、创建 ConnectionFactory 对象，并指定 ActiveMQ 服务IP及端口号（需要指定TCP通信；
    2、使用 ConnectionFactory 对象创建一个 Connection 对象；
    3、开启连接，调用 Connection 对象的 start() 方法；
    4、使用 Connection 对象创建一个 Session 对象；
    5、使用 Session 对象创建一个 Destination 对象（topic、queue），此处创建一个 Queue 对象；
    6、使用 Session 对象构建一个 Producer 对象；
    7、创建 TextMessage 对象，并封装消息；
    8、使用 Producer 对象发送消息；
    9、释放连接资源；

示例代码如下：

package com.huazai.activemq.queue;
    
    import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.MessageProducer;
    import javax.jms.Queue;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    import org.junit.Test;
    
    /**
     * 
     * @author HuaZai
     * @contact who.seek.me@java98k.vip
     *          <ul>
     * @description ActiveMQ的Queue的Producer测试单元
     *              </ul>
     * @className ActiveMQTest
     * @package com.huazai.activemq.queue
     * @createdTime 2017年06月18日
     *
     * @version V1.0.0
     */
    public class TQueueProducer
    {
    
    	@Test
    	public void test() throws Exception
    	{
    		// 1、创建 ConnectionFactory 对象，并指定 ActiveMQ 服务IP及端口号（需要指定TCP通信）
    		ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://192.168.194.131:61616");
    		// 2、使用 ConnectionFactory 对象创建一个 Connection 对象
    		Connection connection = factory.createConnection();
    		// 3、开启连接，调用 Connection 对象的 start() 方法
    		connection.start();
    		// 4、使用 Connection 对象创建一个 Session 对象
    		// 第一个参数：表示是否开启分布式事务（JTA） 一般就是false :表示不开启。 只有设置了false ,第二个参数才有意义。
    		// 第二个参数：表示设置应答模式 自动应答和手动应答 。一般使用的是自动应答
    		Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    		// 5、使用 Session 对象创建一个 Destination 对象（topic、queue），此处创建一个 Queue 对象
    		Queue query = session.createQueue("queue-001");
    		// 6、使用 Session 对象构建一个 Producer 对象
    		MessageProducer producer = session.createProducer(query);
    		// 7、创建 TextMessage 对象，并封装消息
    		TextMessage message = session.createTextMessage("Hello test ActiveMQ queue!!!");
    		// 8、使用 Producer 对象发送消息
    		producer.send(message);
    		// 9、释放连接资源
    		session.close();
    		producer.close();
    		connection.close();
    	}
    }

Consumer 创建步骤（负责消费消息）：

1、创建 ConnectionFactory 对象，并指定 ActiveMQ 服务IP及端口号（需要指定TCP通信）；
    2、使用 ConnectionFactory 对象创建一个 Connection 对象；
    3、开启连接，调用 Connection 对象的 start() 方法；
    4、使用 Connection 对象创建一个 Session 对象；
    5、使用 Session 对象创建一个 Destination 对象（topic、queue），此处创建一个 Queue 对象；
    6、使用 Session 对象构建一个 Consumer 对象；
    7、接收消息，并解析输出；
    8、关闭资源；

示例代码如下：

package com.huazai.activemq.queue;
    
    import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageConsumer;
    import javax.jms.MessageListener;
    import javax.jms.Queue;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    import org.junit.Test;
    
    /**
     * 
     * @author HuaZai
     * @contact who.seek.me@java98k.vip
     *          <ul>
     * @description ActiveMQ的Queue的Consumer测试单元
     *              </ul>
     * @className ActiveMQTest
     * @package com.huazai.activemq.queue
     * @createdTime 2017年06月18日
     *
     * @version V1.0.0
     */
    public class TQueueConsumer {
    	
    	@Test
    	public void recieve() throws Exception {
    		// 1、创建 ConnectionFactory 对象，并指定 ActiveMQ 服务IP及端口号（需要指定TCP通信）
    		ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://192.168.194.131:61616");
    		// 2、使用 ConnectionFactory 对象创建一个 Connection 对象
    		Connection connection = factory.createConnection();
    		// 3、开启连接，调用 Connection 对象的 start() 方法
    		connection.start();
    		// 4、使用 Connection 对象创建一个 Session 对象
    		// 第一个参数：表示是否开启分布式事务（JTA） 一般就是false :表示不开启。 只有设置了false ,第二个参数才有意义。
    		// 第二个参数：表示设置应答模式 自动应答和手动应答 。使用的是自动应答
    		Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    		// 5、使用 Session 对象创建一个 Destination 对象（topic、queue），此处创建一个 Queue 对象
    		Queue query = session.createQueue("queue-001");
    		// 6、使用 Session 对象构建一个 Consumer 对象
    		MessageConsumer consumer = session.createConsumer(query);
    		// 7、接收消息，并解析输出
    		
    		//第一种接收消息.直接接收  只是测试的使用
    		/*while(true){
    			//设置接收消息的超时时间 单位是毫秒
    			Message receive = consumer.receive(3000000);
    			
    			if(receive==null){
    				break;
    			}
    			
    			//取消息
    			if(receive instanceof TextMessage){
    				TextMessage message = (TextMessage)receive;
    				String text = message.getText();//获取消息的内容
    				System.out.println(text);
    			}
    		}*/
    		
    		//第二种接收消息.设置一个监听器  就是开启了一个新的线程
    		System.out.println("start");
    		consumer.setMessageListener(new MessageListener() {
    			
    			@Override
    			public void onMessage(Message message) {
    				if(message instanceof TextMessage){
    					TextMessage message2 = (TextMessage)message;
    					String text="";
    					try {
    						text = message2.getText();
    					} catch (JMSException e) {
    						e.printStackTrace();
    					}
    					//输出消息内容
    					System.out.println(text);
    				}
    			}
    		});
    		System.out.println("end");
    		// 模拟消费者挂起
    		Thread.sleep(10000000);
    		
    		// 8、关闭资源
    		consumer.close();
    		session.close();
    		connection.close();
    	}
    }

2）、队列（Queue）的使用

Producer 创建步骤（负责生产消息）：

同上。。。

示例代码如下：

package com.huazai.activemq.topic;
    
    import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.MessageProducer;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    import javax.jms.Topic;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    import org.junit.Test;
    
    /**
     * 
     * @author HuaZai
     * @contact who.seek.me@java98k.vip
     *          <ul>
     * @description ActiveMQ的Topic的Producer测试单元
     *              </ul>
     * @className TopicProducer
     * @package com.huazai.activemq.topic
     * @createdTime 2017年06月18日
     *
     * @version V1.0.0
     */
    public class TopicProducer {
    
    	@Test
    	public void send() throws Exception {
    		// 1、创建 ConnectionFactory 对象，并指定 ActiveMQ 服务 IP 及端口号（需要指定TCP通信）
    		ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://192.168.194.131:61616");
    		// 2、使用 ConnectionFactory 对象创建一个 Connection 对象
    		Connection connection = factory.createConnection();
    		// 3、开启连接，调用 Connection 对象的start方法
    		connection.start();
    		// 4、使用 Connection 对象创建一个 Session 对象
    		// 第一个参数：表示是否开启分布式事务（JTA） 一般就是false :表示不开启。 只有设置了false ,第二个参数才有意义。
    		// 第二个参数：表示设置应答模式 自动应答和手动应答 。使用的是自动应答
    		Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    		// 5、使用 session 创建目的地（destination）topic
    		Topic topic = session.createTopic("topic-001");
    		// 6、使用 Session 对象创建一个 Producer 对象
    		MessageProducer producer = session.createProducer(topic);
    		// 7、创建 TextMessage 对象，并封装消息
    		TextMessage message = session.createTextMessage("Hello test ActiveMQ topic!!!");
    		// 8、使用 Producer 对象发送消息
    		producer.send(message);
    		// 9、释放连接资源
    		session.close();
    		producer.close();
    		connection.close();
    	}
    }

Consumer 创建步骤（负责消费消息）：

创建步骤同上。。。

示例代码如下：

订阅者1，

package com.huazai.activemq.topic;
    
    import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageConsumer;
    import javax.jms.MessageListener;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    import javax.jms.Topic;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    import org.junit.Test;
    
    /**
     * 
     * @author HuaZai
     * @contact who.seek.me@java98k.vip
     *          <ul>
     * @description ActiveMQ的Topic的Consumer1测试单元
     *              </ul>
     * @className TopicProducer
     * @package com.huazai.activemq.topic
     * @createdTime 2017年06月18日
     *
     * @version V1.0.0
     */
    public class TopicCustomer1 {
    
    	@Test
    	public void reieve() throws Exception {
    		// 1、创建ConnectionFactory对象，并指定ActiveMQ服务ip及端口号（需要指定TCP通信）
    		ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://192.168.194.131:61616");
    		// 2、使用ConnectionFactory对象创建一个Connection对象
    		Connection connection = factory.createConnection();
    		// 3、开启连接，调用Connection对象的start方法
    		connection.start();
    		// 4、使用Connection对象创建一个Session对象
    		// 第一个参数：表示是否开启分布式事务（JTA） 一般就是false :表示不开启。 只有设置了false ,第二个参数才有意义。
    		// 第二个参数：表示设置应答模式 自动应答和手动应答 。使用的是自动应答
    		Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    		// 5、使用session创建目的地（destination）topic
    		Topic topic = session.createTopic("topic-001");
    		// 6、使用Session对象创建一个Producer对象
    		MessageConsumer consumer = session.createConsumer(topic);
    		// 7.接收消息
    
    		// 设置消息监听器
    		System.out.println("start");
    		consumer.setMessageListener(new MessageListener() {
    
    			@Override
    			public void onMessage(Message message) {
    				if (message instanceof TextMessage) {
    					TextMessage message2 = (TextMessage) message;
    					String text = "";
    					try {
    						text = message2.getText();
    					} catch (JMSException e) {
    						e.printStackTrace();
    					}
    					// 获取消息的内容
    					System.out.println(text);
    				}
    			}
    		});
    		System.out.println("end");
    		// 睡眠
    		Thread.sleep(10000000);
    
    		// 9.、关闭资源
    		consumer.close();
    		session.close();
    		connection.close();
    	}
    }

订阅者2，

package com.huazai.activemq.topic;
    
    import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageConsumer;
    import javax.jms.MessageListener;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    import javax.jms.Topic;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    import org.junit.Test;
    
    /**
     * 
     * @author HuaZai
     * @contact who.seek.me@java98k.vip
     *          <ul>
     * @description ActiveMQ的Topic的Consumer1测试单元
     *              </ul>
     * @className TopicProducer
     * @package com.huazai.activemq.topic
     * @createdTime 2017年06月18日
     *
     * @version V1.0.0
     */
    public class TopicCustomer1 {
    
    	@Test
    	public void reieve() throws Exception {
    		// 1、创建ConnectionFactory对象，并指定ActiveMQ服务ip及端口号（需要指定TCP通信）
    		ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://192.168.194.131:61616");
    		// 2、使用ConnectionFactory对象创建一个Connection对象
    		Connection connection = factory.createConnection();
    		// 3、开启连接，调用Connection对象的start方法
    		connection.start();
    		// 4、使用Connection对象创建一个Session对象
    		// 第一个参数：表示是否开启分布式事务（JTA） 一般就是false :表示不开启。 只有设置了false ,第二个参数才有意义。
    		// 第二个参数：表示设置应答模式 自动应答和手动应答 。使用的是自动应答
    		Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    		// 5、使用session创建目的地（destination）topic
    		Topic topic = session.createTopic("topic-001");
    		// 6、使用Session对象创建一个Producer对象
    		MessageConsumer consumer = session.createConsumer(topic);
    		// 7、接收消息，并输出内容
    
    		// 设置消息监听器
    		System.out.println("start");
    		consumer.setMessageListener(new MessageListener() {
    
    			@Override
    			public void onMessage(Message message) {
    				if (message instanceof TextMessage) {
    					TextMessage message2 = (TextMessage) message;
    					String text = "";
    					try {
    						text = message2.getText();
    					} catch (JMSException e) {
    						e.printStackTrace();
    					}
    					// 获取消息的内容
    					System.out.println(text);
    				}
    			}
    		});
    		System.out.println("end");
    		// 睡眠
    		Thread.sleep(10000000);
    
    		// 9、关闭资源
    		consumer.close();
    		session.close();
    		connection.close();
    	}
    }

注意：

Topic 默认是不存在于MQ服务器中的，一旦发送之后，如果没有订阅，就没了。  
Queue 默认是存在于MQ的服务器中的，发送消息之后，随时取。但是一定是一个消费者取完就没了。

七、ActiveMQ 整合 Spring

配置pom.xml，内容如下：

<project xmlns="http://maven.apache.org/POM/4.0.0"
    	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    	<modelVersion>4.0.0</modelVersion>
    	<parent>
    		<groupId>com.huazai</groupId>
    		<artifactId>huazai-parent</artifactId>
    		<version>0.0.1-SNAPSHOT</version>
    	</parent>
    	<artifactId>huazai-activemq</artifactId>
    	<dependencies>
    		
    		<dependency>
    			<groupId>org.springframework</groupId>
    			<artifactId>spring-context</artifactId>
    		</dependency>
    		<dependency>
    			<groupId>org.springframework</groupId>
    			<artifactId>spring-beans</artifactId>
    		</dependency>
    		<dependency>
    			<groupId>org.springframework</groupId>
    			<artifactId>spring-webmvc</artifactId>
    		</dependency>
    		<dependency>
    			<groupId>org.springframework</groupId>
    			<artifactId>spring-jdbc</artifactId>
    		</dependency>
    		<dependency>
    			<groupId>org.springframework</groupId>
    			<artifactId>spring-aspects</artifactId>
    		</dependency>
    		
    		<dependency>
    			<groupId>org.springframework</groupId>
    			<artifactId>spring-jms</artifactId>
    		</dependency>
    		<dependency>
    			<groupId>org.springframework</groupId>
    			<artifactId>spring-context-support</artifactId>
    		</dependency>
    		<dependency>
    			<groupId>org.apache.activemq</groupId>
    			<artifactId>activemq-all</artifactId>
    		</dependency>
    	</dependencies>
    </project>

新增 Spring 配置文件（applicationContext-activemq.xml）：

<?xml version="1.0" encoding="UTF-8"?>
    <beans xmlns="http://www.springframework.org/schema/beans"
    	xmlns:context="http://www.springframework.org/schema/context"
    	xmlns:p="http://www.springframework.org/schema/p"
    	xmlns:aop="http://www.springframework.org/schema/aop"
    	xmlns:tx="http://www.springframework.org/schema/tx"
    	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.2.xsd
    	http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.2.xsd
    	http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-4.2.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-4.2.xsd
    	http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util-4.2.xsd">
    
    	
    	<bean id="targetConnection"
    		class="org.apache.activemq.ActiveMQConnectionFactory">
    		<property name="brokerURL"
    			value="tcp://192.168.194.131:61616"></property>
    	</bean>
    
    	
    	<bean id="connectionFactory"
    		class="org.springframework.jms.connection.SingleConnectionFactory">
    		<property name="targetConnectionFactory"
    			ref="targetConnection"></property>
    	</bean>
    
    	
    	<bean class="org.springframework.jms.core.JmsTemplate">
    		<property name="connectionFactory" ref="connectionFactory"></property>
    	</bean>
    
    	
    
    	<bean id="topicDestination"
    		class="org.apache.activemq.command.ActiveMQTopic">
    		<constructor-arg name="name" value="spring-topic-001"></constructor-arg>
    	</bean>
    
    	
    	<bean id="activeMQMessageListener"
    		class="com.huazai.activemq.spring.ActiveMQMessageListener"></bean>
    	<bean
    		class="org.springframework.jms.listener.DefaultMessageListenerContainer">
    		<property name="connectionFactory" ref="connectionFactory"></property>
    		<property name="destination" ref="topicDestination"></property>
    		<property name="messageListener"
    			ref="activeMQMessageListener"></property>
    	</bean>
    </beans>

创建消息生产者，示例代码如下：

package com.huazai.activemq.spring;
    
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.Session;
    
    import org.apache.xbean.spring.context.ClassPathXmlApplicationContext;
    import org.junit.Test;
    import org.springframework.context.ApplicationContext;
    import org.springframework.jms.core.JmsTemplate;
    import org.springframework.jms.core.MessageCreator;
    
    /**
     * 
     * @author HuaZai
     * @contact who.seek.me@java98k.vip
     *          <ul>
     * @description ActiveMQ的Producer测试单元
     *              </ul>
     * @className TopicProducer
     * @package com.huazai.activemq.spring
     * @createdTime 2017年06月18日
     *
     * @version V1.0.0
     */
    public class Producer {
    
    	private ApplicationContext context;
    
    	@Test
    	public void send() throws Exception {
    		// 1、初始化Spring容器
    		context = new ClassPathXmlApplicationContext("classpath:applicationContext-activemq.xml");
    		// 2、获取JmsTemplate对象
    		JmsTemplate jmsTemplate = context.getBean(JmsTemplate.class);
    		// 3、获取Destination对象
    		Destination destination = (Destination) context.getBean(Destination.class);
    		// 4、发送消息
    		jmsTemplate.send(destination, new MessageCreator() {
    
    			@Override
    			public Message createMessage(Session session) throws JMSException {
    				return session.createTextMessage("Hello test Spring With ActiveMQ Send Message!!!");
    			}
    		});
    		Thread.sleep(100000);
    	}
    }

创建消息监听对象（MessageListener），示例代码如下：

package com.huazai.activemq.spring;
    
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageListener;
    import javax.jms.TextMessage;
    
    /**
     * 
     * @author HuaZai
     * @contact who.seek.me@java98k.vip
     *          <ul>
     * @description ActiveMQ的MessageListener
     *              </ul>
     * @className TopicProducer
     * @package com.huazai.activemq.spring
     * @createdTime 2017年06月18日
     *
     * @version V1.0.0
     */
    public class ActiveMQMessageListener implements MessageListener {
    
    	@Override
    	public void onMessage(Message message) {
    		// 接收消息
    		if (message instanceof TextMessage) {
    			TextMessage textMessage = (TextMessage) message;
    			String text;
    			try {
    				text = textMessage.getText();
    				// 输出消息内容
    				System.out.println(text);
    			} catch (JMSException e) {
    				e.printStackTrace();
    			}
    		}
    	}
    
    }

源码地址：【[GitHub\_huazai-activemq][GitHub_huazai-activemq]】

--------------------

好了，关于 ActiveMQ 进阶与应用详解 就写到这儿了，如果还有什么疑问或遇到什么问题欢迎扫码提问，也可以给我留言哦，我会一一详细的解答的。   
歇后语：“ 共同学习，共同进步 ”，也希望大家多多关注CSND的IT社区。

--------------------

<table> 
 <tbody> 
  <tr> 
   <td>作&nbsp; &nbsp; &nbsp; &nbsp;者：</td> 
   <td>华 &nbsp; &nbsp;仔</td> 
  </tr> 
  <tr> 
   <td>联系作者：</td> 
   <td>who.seek.me@java98k.vip</td> 
  </tr> 
  <tr> 
   <td>来&nbsp; &nbsp; &nbsp; &nbsp; 源：</td> 
   <td>CSDN (Chinese Software Developer Network)</td> 
  </tr> 
  <tr> 
   <td>原&nbsp; &nbsp; &nbsp; &nbsp; 文：</td> 
   <td><a href="https://blog.csdn.net/Hello_World_QWP/article/details/103187961" rel="nofollow">https://blog.csdn.net/Hello_World_QWP/article/details/103187961</a></td> 
  </tr> 
  <tr> 
   <td>版权声明：</td> 
   <td>本文为博主原创文章，请在转载时务必注明博文出处！</td> 
  </tr> 
 </tbody> 
</table>

[Linux_Jdk1.8]: https://blog.csdn.net/Hello_World_QWP/article/details/78861247
[activemq.apache]: http://activemq.apache.org/
[20191202104629499.png]: https://img-blog.csdnimg.cn/20191202104629499.png
[20191204110555574.png]: https://img-blog.csdnimg.cn/20191204110555574.png
[http_192.168.159.129_8161_index.html]: http://192.168.159.129:8161/index.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0hlbGxvX1dvcmxkX1FXUA_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20191204111418232.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0hlbGxvX1dvcmxkX1FXUA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0hlbGxvX1dvcmxkX1FXUA_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20191204111620222.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0hlbGxvX1dvcmxkX1FXUA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0hlbGxvX1dvcmxkX1FXUA_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20191204114754321.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0hlbGxvX1dvcmxkX1FXUA==,size_16,color_FFFFFF,t_70
[GitHub_huazai-activemq]: https://github.com/Jackson-AndyLau/huazai-activemq