ActiveMQ环境搭建及实例详解

雨点打透心脏的1/2处 2022-06-05 03:26 711阅读 0赞

# ActiveMQ环境搭建及实例详解 #

> 前言：  
> JMS（Java Messaging Service）是Java平台上有关面向消息中间件的技术规范，它便于消息系统中的Java应用程序进行消息交换,并且通过提供标准的产生、发送、接收消息的接口简化企业应用的开发。而ActiveMQ是这个规范的一个具体实现。

**ActiveMQ是什么？：ActiveMQ是一个消息中间件的一种**  
**ActiveMQ干什么？：他提供标准的产生、发送、接收消息接口**  
**ActiveMQ怎么用？：ＪＡＶＡ开发的时候就可以直接实现相关的接口，完成ＪＡＶＡ应用程序之间的消息发送。**

## 1、ActiveMQ安装部署(Windows) ##

> 安装包下载链接：  
> [http://pan.baidu.com/s/1dF71zGH ][http_pan.baidu.com_s_1dF71zGH]  
> 密码：4tv3

这是一个免安装版的，解压后，**双击下图的文件**，就可以启动ActiveMQ这个消息中间件）（相当于一个服务器），当我们编辑好相关接口的实现类，发送消息，就能实时接收到消息了：

![这里写图片描述][SouthEast]

启动后的界面：  
![这里写图片描述][SouthEast 1]

** 代码详情：**

wrapper  | --> Wrapper Started as Console
    wrapper  | Launching a JVM...
    jvm 1    | Wrapper (Version 3.2.3) http://wrapper.tanukisoftware.org
    jvm 1    |   Copyright 1999-2006 Tanuki Software, Inc.  All Rights Reserved.
    jvm 1    |
    jvm 1    | Java Runtime: Oracle Corporation 1.8.0_144 C:\Program Files\Java\jre1.8.0_144
    jvm 1    |   Heap sizes: current=251392k  free=235660k  max=932352k
    jvm 1    |     JVM args: -Dactivemq.home=../.. -Dactivemq.base=../.. -Djavax.net.ssl.keyStorePassword=password -Djavax.net.ssl.trustStorePassword=password -Djavax.net.ssl.keyStore=../../conf/broker.ks -Djavax.net.ssl.trustStore=../../conf/broker.ts -Dcom.sun.management.jmxremote -Dorg.apache.activemq.UseDedicatedTaskRunner=true -Djava.util.logging.config.file=logging.properties -Dactivemq.conf=../../conf -Dactivemq.data=../../data -Djava.security.auth.login.config=../../conf/login.config -Xmx1024m -Djava.library.path=../../bin/win64 -Dwrapper.key=5q7ouaKXTUiBhUBy -Dwrapper.port=32000 -Dwrapper.jvm.port.min=31000 -Dwrapper.jvm.port.max=31999 -Dwrapper.pid=16200 -Dwrapper.version=3.2.3 -Dwrapper.native_library=wrapper -Dwrapper.cpu.timeout=10 -Dwrapper.jvmid=1
    jvm 1    | Extensions classpath:
    jvm 1    |   [..\..\lib,..\..\lib\camel,..\..\lib\optional,..\..\lib\web,..\..\lib\extra]
    jvm 1    | ACTIVEMQ_HOME: ..\..
    jvm 1    | ACTIVEMQ_BASE: ..\..
    jvm 1    | ACTIVEMQ_CONF: ..\..\conf
    jvm 1    | ACTIVEMQ_DATA: ..\..\data
    jvm 1    | Loading message broker from: xbean:activemq.xml
    jvm 1    |  INFO | Refreshing org.apache.activemq.xbean.XBeanBrokerFactory$1@5ea3ef68: startup date [Thu Nov 16 17:38:15 CST 2017]; root of context hierarchy
    jvm 1    |  INFO | PListStore:[D:\apache-activemq-5.11.2\bin\win64\..\..\data\localhost\tmp_storage] started
    jvm 1    |  INFO | Using Persistence Adapter: KahaDBPersistenceAdapter[D:\apache-activemq-5.11.2\bin\win64\..\..\data\kahadb]
    jvm 1    |  INFO | KahaDB is version 5
    jvm 1    |  INFO | Recovering from the journal ...
    jvm 1    |  INFO | Recovery replayed 5 operations from the journal in 0.022 seconds.
    jvm 1    |  INFO | Apache ActiveMQ 5.11.2 (localhost, ID:DESKTOP-TKN3IRE-49495-1510825096880-0:1) is starting
    jvm 1    |  INFO | Listening for connections at: tcp://DESKTOP-TKN3IRE:61616?maximumConnections=1000&wireFormat.maxFrameSize=104857600
    jvm 1    |  INFO | Connector openwire started
    jvm 1    |  INFO | Listening for connections at: amqp://DESKTOP-TKN3IRE:5672?maximumConnections=1000&wireFormat.maxFrameSize=104857600
    jvm 1    |  INFO | Connector amqp started
    jvm 1    |  INFO | Listening for connections at: stomp://DESKTOP-TKN3IRE:61613?maximumConnections=1000&wireFormat.maxFrameSize=104857600
    jvm 1    |  INFO | Connector stomp started
    jvm 1    |  INFO | Listening for connections at: mqtt://DESKTOP-TKN3IRE:1883?maximumConnections=1000&wireFormat.maxFrameSize=104857600
    jvm 1    |  INFO | Connector mqtt started
    jvm 1    | {}
    jvm 1    |  INFO | Listening for connections at ws://DESKTOP-TKN3IRE:61614?maximumConnections=1000&wireFormat.maxFrameSize=104857600
    jvm 1    |  INFO | Connector ws started
    jvm 1    |  INFO | Apache ActiveMQ 5.11.2 (localhost, ID:DESKTOP-TKN3IRE-49495-1510825096880-0:1) started
    jvm 1    |  INFO | For help or more information please see: http://activemq.apache.org
    jvm 1    |  INFO | ActiveMQ WebConsole available at http://0.0.0.0:8161/
    jvm 1    |  INFO | Initializing Spring FrameworkServlet 'dispatcher'
    jvm 1    |  INFO | jolokia-agent: No access restrictor found at classpath:/jolokia-access.xml, access to all MBeans is allowed

注：请读者自己解析打印的信息，很重要，也不难。

### 1.1、ActiveMQ控制台界面 ###

ActiveMQ默认启动到8161端口，启动完了后在浏览器地址栏输入：[http://localhost:8161/admin要求输入用户名密码，默认用户名密码为admin、admin，这个用户名密码是在][http_localhost_8161_admin_admin_admin]**conf/users.properties**中配置的。输入用户名密码后便可看到如下图的ActiveMQ控制台界面了。

![这里写图片描述][SouthEast 2]

### 1.2、ActiveMQ目录结构 ###

![这里写图片描述][SouthEast 3]

> 目录结构解析：  
>      bin:存放的是脚本文件  
>      conf:存放的是基本配置文件  
>      data:存放的是日志文件  
>      docs:存放的是说明文档  
>      examples:存放的是简单的实例  
>      lib:存放的是activemq所需jar包  
>      webapps:用于存放项目的目录

## 2、ActiveMQ模型 ##

前面我们已经搭建和配置好了ActiveMQ,下面来看一个Demo,体验一下MQ。JMS消息服务应用程序结构支持两种模型：点对点模型，发布者/订阅者模型。

### 2.1、点对点模型(Queue) ###

一个生产者向一个特定的队列发布消息，一个消费者从这个队列中依次读取消息。  
　模型特点：只有一个消费者获得消息。  
　  
![这里写图片描述][SouthEast 4]

### 2.2、发布者/订阅者模型（Topic） ###

0个或多个订阅者可以接受特定主题的消息。  
　模型特点：多个消费者可获得消息。  
　  
　![这里写图片描述][SouthEast 5]　　  
　　　　  
   Topic和Queue的最大区别在于Topic是以广播的形式，通知所有在线监听的客户端有新的消息，没有监听的客户端将收不到消息；而Queue则是以点对点的形式通知多个处于监听状态的客户端中的一个。

> JMS五种消息格式：  
>        MapMessage ：key-value键值对  
>        TextMessage ：字符串对象  
>        ObjcetMessage ： 一个序列化的Java对象  
>       ByteMessage ：一个未解释字节的数据流  
>        StreamMessage ： Java原始值的数据流

### 2.3、ActiveMQ开发流程 ###

![这里写图片描述][SouthEast 6]

> 1、根据用户名、密码、连接地址获取ActiveMQConnectionFactory对象  
> 2、由连接工厂获取创建连接Connection  
> 4、由连接对象创建会话Session（重要）  
> 5、由会话对象Session根据实际需要创建消息目的地Destination、消息生产者MessageProducer、消息消费者MessageConsumer、消息格式类型（如：TextMessage）  
> 6、发送消息： messageProducer.send(message);  
> 7、接受消息：messageConsumer.receive(100000);

## 3、ActiveMQ使用之Queue（点对点消息实现） ##

如下图，创建一个JAVA项目，引入activemq-all-5.11.1.jar包：  
                       ![这里写图片描述][SouthEast 7]

### 3.1、创建消息生产者（Queue）： ###

package Producer;
    
    import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.MessageProducer;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    
    import org.apache.activemq.ActiveMQConnection;
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    /**
     * 消息生产者 
     * @author yang
     *
     */
    public class JMSProducer {
    
    	private static final String USERNAME=ActiveMQConnection.DEFAULT_USER; // 默认的连接用户名  
    	private static final String PASSWORD=ActiveMQConnection.DEFAULT_PASSWORD; // 默认的连接密码  
    	private static final String BROKEURL=ActiveMQConnection.DEFAULT_BROKER_URL; // 默认的连接地址  
    	private static final int SENDNUM=10; // 发送的消息数量  
    	
    	 /** 
         * 发送消息    静态类
         * @param session        会话对象，创建本次会话的基本配置
         * @param messageProducer 消息生产者对象，将本次消息发送给Queue
         * @throws Exception 
         */  
        public static void sendMessage(Session session,MessageProducer messageProducer)throws Exception{  
            for(int i=0;i<JMSProducer.SENDNUM;i++){  
            	//创建消息的格式（以这种格式发送）： TextMessage
                TextMessage message=session.createTextMessage("ActiveMQ 发送过来的消息"+i+"第二次"); 
                //后台打印
                System.out.println("发送消息："+"ActiveMQ 发送过来的消息"+i+"第二次");  
                //将消息发送给Queue
                messageProducer.send(message);  
            }  
        }  
    	 
    	public static void main(String[] args) {
    		ConnectionFactory connectionFactory; // 连接工厂  
            Connection connection = null; // 连接  
            Session session; // 会话 接受或者发送消息的线程  
            Destination destination; // 消息的目的地  
            MessageProducer messageProducer; // 消息生产者  
              
            // 实例化连接工厂  
            connectionFactory=new ActiveMQConnectionFactory(JMSProducer.USERNAME, JMSProducer.PASSWORD, JMSProducer.BROKEURL);  
              
            try {  
                connection=connectionFactory.createConnection(); // 通过连接工厂获取连接  
                connection.start(); // 启动连接  
                session=connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); // 创建Session  
                destination=session.createQueue("FirstQueue1"); // 创建消息队列  
                messageProducer=session.createProducer(destination); // 创建消息生产者  
                // 发送消息 ，本质上是将TextMessage类型的消息发送给Queue这个消息队列，这是消息的目的地
                sendMessage(session, messageProducer);
                session.commit();  
            } catch (Exception e) {    
                e.printStackTrace();  
            } finally{  
                if(connection!=null){  
                    try {  
                        connection.close();  
                    } catch (JMSException e) {                    
                        e.printStackTrace();  
                    }  
                }  
            }  
    	}
    	
    	
    	
    }

### 3.2、消息接收者： ###

package Consumer;
    
    import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.MessageConsumer;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    
    import org.apache.activemq.ActiveMQConnection;
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    /**
     * 消息消费者
     * @author yang
     *
     */
    public class JMSConsumer {
    
    	private static final String USERNAME=ActiveMQConnection.DEFAULT_USER; // 默认的连接用户名  
    	private static final String PASSWORD=ActiveMQConnection.DEFAULT_PASSWORD; // 默认的连接密码  
    	private static final String BROKEURL=ActiveMQConnection.DEFAULT_BROKER_URL; // 默认的连接地址  
    	
    	public static void main(String[] args) {
    		ConnectionFactory connectionFactory; // 连接工厂  
            Connection connection = null; // 连接  
            Session session; // 会话 接受或者发送消息的线程  
            Destination destination; // 消息的目的地  
            MessageConsumer messageConsumer; // 消息的消费者  
              
            // 实例化连接工厂  
            connectionFactory=new ActiveMQConnectionFactory(JMSConsumer.USERNAME, JMSConsumer.PASSWORD, JMSConsumer.BROKEURL);  
                      
            try {  
                connection=connectionFactory.createConnection();  // 通过连接工厂获取连接  
                connection.start(); // 启动连接  
                // 创建Session
                session=connection.createSession(Boolean.FALSE,                  Session.AUTO_ACKNOWLEDGE);  
                // 创建连接的消息队列  （我们消费者要去哪个队列上读取消息） 
                destination=session.createQueue("FirstQueue1");  
                // 创建消息消费者  
                messageConsumer=session.createConsumer(destination); 
                while(true){  
                    //取出队列中  TextMessage格式的消息
                    TextMessage textMessage=(TextMessage)messageConsumer.receive(100000);
                    if(textMessage!=null){  
                        System.out.println("收到的消息："+textMessage.getText());//消息格式转化+打印  
                    }else{  
                        break;  
                    }  
                }  
            } catch (JMSException e) {    
                e.printStackTrace();  
            }   
    
    	}
    
    }

### 3.3、启动ActiveMQ ###

#### 3.3.1、双击下面图标： ####

![这里写图片描述][SouthEast 8]

#### 3.3.2、出现下面的界面： ####

![这里写图片描述][SouthEast 9]

\***表示ActiveMQ的队列（Queue）开始工作，能够接受来自生产者的消息，消费者能够从队列中获得消息。**

#### 3.3.2、启动消息生产者： ####

![这里写图片描述][SouthEast 10]  
\*消息已经发送到Queue中

#### 3.3.2、启动消息消费者： ####

![这里写图片描述][SouthEast 11]

## 4、ActiveMQ使用之Queue（引入监听点对点消息） ##

使用上面的方式有一个弊端：需要，每次操作消费者才能收到消息，使用监听后，消费者实时监听Queue，当生产者发送消息时，就能接受到消息。

### 4.1、创建一个监听对象 ###

package Listener;
    
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageListener;
    import javax.jms.TextMessage;
    /**
     *  监听器类：
     * @author yang
     *
     */
    public class Listener implements MessageListener {
    
    	@Override
    	public void onMessage(Message message) {
    		try {  
                System.out.println("收到的消息："+((TextMessage)message).getText());  
            } catch (JMSException e) {  
                e.printStackTrace();  
            }  
    
    	}
    
    }

### 4.2、创建一个消费者，引入监听对象 ###

package Consumer;
    
    import javax.jms.Connection;  
    import javax.jms.ConnectionFactory;  
    import javax.jms.Destination;  
    import javax.jms.JMSException;  
    import javax.jms.MessageConsumer;  
    import javax.jms.Session;  
    import javax.jms.TextMessage;  
      
    import org.apache.activemq.ActiveMQConnection;  
    import org.apache.activemq.ActiveMQConnectionFactory;  
    
    import Listener.Listener;
    /**
     * 消息消费者中注册监听器：
     * @author yang
     *
     */
    public class JMSConsumer2 {
    
    	private static final String USERNAME=ActiveMQConnection.DEFAULT_USER; // 默认的连接用户名  
    	private static final String PASSWORD=ActiveMQConnection.DEFAULT_PASSWORD; // 默认的连接密码  
    	private static final String BROKEURL=ActiveMQConnection.DEFAULT_BROKER_URL; // 默认的连接地址  
    	      
    	
    	
    	public static void main(String[] args) {
    		ConnectionFactory connectionFactory; // 连接工厂  
            Connection connection = null; // 连接  
            Session session; // 会话 接受或者发送消息的线程  
            Destination destination; // 消息的目的地  
            MessageConsumer messageConsumer; // 消息的消费者  
              
            // 实例化连接工厂  
            connectionFactory=new ActiveMQConnectionFactory(JMSConsumer2.USERNAME, JMSConsumer2.PASSWORD, JMSConsumer2.BROKEURL);  
                      
            try {  
                connection=connectionFactory.createConnection();  // 通过连接工厂获取连接  
                connection.start(); // 启动连接  
                session=connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); // 创建Session  
                destination=session.createQueue("FirstQueue1");  // 创建连接的消息队列  
                messageConsumer=session.createConsumer(destination); // 创建消息消费者  
                messageConsumer.setMessageListener(new Listener()); // 注册消息监听  
            } catch (JMSException e) {   
                e.printStackTrace();  
            }   
    
    	}
    
    }

### 4.3、启动消费者，实时监听Queue ###

![这里写图片描述][SouthEast 12]

### 4.4、多次启动生产者，发送消息 ###

![这里写图片描述][SouthEast 13]

### 4.5、消费者自动获得消息 ###

![这里写图片描述][SouthEast 14]

### 4.5、消费者第二次自动获得消息 ###

![这里写图片描述][SouthEast 15]

![这里写图片描述][SouthEast 16]

## 5、ActiveMQ 使用之Topic（发布-订阅消息模式实现） ##

发布订阅模式：发布者发布消息后，只要订阅了该发布者的订阅者，都能够接受到数据，因此，要想接收数据，必须先要订阅。一个生产者发布对应多个消费者订阅。  
    接着上面的项目，我们再新增一个监听器，一个消费者，项目如下图所示：

![这里写图片描述][SouthEast 17]

\#\#\#5.1、消息生产者（Topic）

package Producer;
    
    import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.MessageProducer;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    
    import org.apache.activemq.ActiveMQConnection;
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    /**
     * 消息生产者 
     * @author yang
     *
     */
    public class JMSProducer2 {
    
    	private static final String USERNAME=ActiveMQConnection.DEFAULT_USER; // 默认的连接用户名  
    	private static final String PASSWORD=ActiveMQConnection.DEFAULT_PASSWORD; // 默认的连接密码  
    	private static final String BROKEURL=ActiveMQConnection.DEFAULT_BROKER_URL; // 默认的连接地址  
    	private static final int SENDNUM=10; // 发送的消息数量  
    	
    	 /** 
         * 发送消息    静态类
         * @param session        会话对象，创建本次会话的基本配置
         * @param messageProducer 消息生产者对象，将本次消息发送给Queue
         * @throws Exception 
         */  
        public static void sendMessage(Session session,MessageProducer messageProducer)throws Exception{  
            for(int i=0;i<JMSProducer2.SENDNUM;i++){  
            	//创建消息的格式（以这种格式发送）： TextMessage
                TextMessage message=session.createTextMessage("ActiveMQ 发送过来的消息"+i+"第三次"); 
                //后台打印
                System.out.println("发送消息："+"ActiveMQ 发送过来的消息"+i+"第三次");  
                //将消息发送给Queue
                messageProducer.send(message);  
            }  
        }  
    	 
    	public static void main(String[] args) {
    		ConnectionFactory connectionFactory; // 连接工厂  
            Connection connection = null; // 连接  
            Session session; // 会话 接受或者发送消息的线程  
            Destination destination; // 消息的目的地  
            MessageProducer messageProducer; // 消息生产者  
              
            // 实例化连接工厂  
            connectionFactory=new ActiveMQConnectionFactory(JMSProducer2.USERNAME, JMSProducer2.PASSWORD, JMSProducer2.BROKEURL);  
              
            try {  
                connection=connectionFactory.createConnection(); // 通过连接工厂获取连接  
                connection.start(); // 启动连接  
                session=connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); // 创建Session  
                destination=session.createTopic("FirstTopic1");//创建消息发布
                messageProducer=session.createProducer(destination); // 创建消息生产者  
                // 发送消息 ，本质上是将TextMessage类型的消息发送给Queue这个消息队列，这是消息的目的地
                sendMessage(session, messageProducer);
                session.commit();  
            } catch (Exception e) {    
                e.printStackTrace();  
            } finally{  
                if(connection!=null){  
                    try {  
                        connection.close();  
                    } catch (JMSException e) {                    
                        e.printStackTrace();  
                    }  
                }  
            }  
    	}
    	
    	
    	
    }

### 5.2、创建二个监听 ###

package Listener;
    
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageListener;
    import javax.jms.TextMessage;
    /**
     *  监听器类：
     * @author yang
     *
     */
    public class Listener implements MessageListener {
    
    	@Override
    	public void onMessage(Message message) {
    		try {  
                System.out.println("第一个消费者收到的消息："+((TextMessage)message).getText());  
            } catch (JMSException e) {  
                e.printStackTrace();  
            }  
    
    	}
    
    }

package Listener;
    
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageListener;
    import javax.jms.TextMessage;
    /**
     *  监听器类：
     * @author yang
     *
     */
    public class Listener2 implements MessageListener {
    
    	@Override
    	public void onMessage(Message message) {
    		try {  
                System.out.println("第二个消费者收到的消息："+((TextMessage)message).getText());  
            } catch (JMSException e) {  
                e.printStackTrace();  
            }  
    
    	}
    
    }

### 5.3、创建二个消费者 ###

package Consumer;
    
    import javax.jms.Connection;  
    import javax.jms.ConnectionFactory;  
    import javax.jms.Destination;  
    import javax.jms.JMSException;  
    import javax.jms.MessageConsumer;  
    import javax.jms.Session;  
    import javax.jms.TextMessage;  
      
    import org.apache.activemq.ActiveMQConnection;  
    import org.apache.activemq.ActiveMQConnectionFactory;  
    
    import Listener.Listener;
    /**
     * 消息消费者中注册监听器：
     * @author yang
     *
     */
    public class JMSConsumer2 {
    
    	private static final String USERNAME=ActiveMQConnection.DEFAULT_USER; // 默认的连接用户名  
    	private static final String PASSWORD=ActiveMQConnection.DEFAULT_PASSWORD; // 默认的连接密码  
    	private static final String BROKEURL=ActiveMQConnection.DEFAULT_BROKER_URL; // 默认的连接地址  
    	      
    	
    	
    	public static void main(String[] args) {
    		ConnectionFactory connectionFactory; // 连接工厂  
            Connection connection = null; // 连接  
            Session session; // 会话 接受或者发送消息的线程  
            Destination destination; // 消息的目的地  
            MessageConsumer messageConsumer; // 消息的消费者  
              
            // 实例化连接工厂  
            connectionFactory=new ActiveMQConnectionFactory(JMSConsumer2.USERNAME, JMSConsumer2.PASSWORD, JMSConsumer2.BROKEURL);  
                      
            try {  
                connection=connectionFactory.createConnection();  // 通过连接工厂获取连接  
                connection.start(); // 启动连接  
                session=connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); // 创建Session  
                //destination=session.createQueue("FirstQueue1");  // 创建连接的消息队列  
                destination=session.createTopic("FirstTopic1"); //创建连接的消息主题
                messageConsumer=session.createConsumer(destination); // 创建消息消费者  
                messageConsumer.setMessageListener(new Listener()); // 注册消息监听  
            } catch (JMSException e) {   
                e.printStackTrace();  
            }   
    
    	}
    
    }

package Consumer;
    
    import javax.jms.Connection;  
    import javax.jms.ConnectionFactory;  
    import javax.jms.Destination;  
    import javax.jms.JMSException;  
    import javax.jms.MessageConsumer;  
    import javax.jms.Session;  
    import javax.jms.TextMessage;  
      
    import org.apache.activemq.ActiveMQConnection;  
    import org.apache.activemq.ActiveMQConnectionFactory;  
    
    import Listener.Listener;
    import Listener.Listener2;
    /**
     * 消息消费者中注册监听器：
     * @author yang
     *
     */
    public class JMSConsumer3 {
    
    	private static final String USERNAME=ActiveMQConnection.DEFAULT_USER; // 默认的连接用户名  
    	private static final String PASSWORD=ActiveMQConnection.DEFAULT_PASSWORD; // 默认的连接密码  
    	private static final String BROKEURL=ActiveMQConnection.DEFAULT_BROKER_URL; // 默认的连接地址  
    	      
    	
    	
    	public static void main(String[] args) {
    		ConnectionFactory connectionFactory; // 连接工厂  
            Connection connection = null; // 连接  
            Session session; // 会话 接受或者发送消息的线程  
            Destination destination; // 消息的目的地  
            MessageConsumer messageConsumer; // 消息的消费者  
              
            // 实例化连接工厂  
            connectionFactory=new ActiveMQConnectionFactory(JMSConsumer3.USERNAME, JMSConsumer3.PASSWORD, JMSConsumer3.BROKEURL);  
                      
            try {  
                connection=connectionFactory.createConnection();  // 通过连接工厂获取连接  
                connection.start(); // 启动连接  
                session=connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); // 创建Session  
                //destination=session.createQueue("FirstQueue1");  // 创建连接的消息队列  
                destination=session.createTopic("FirstTopic1");  //创建连接的消息主题
                messageConsumer=session.createConsumer(destination); // 创建消息消费者  
                messageConsumer.setMessageListener(new Listener2()); // 注册消息监听  
            } catch (JMSException e) {   
                e.printStackTrace();  
            }   
    
    	}
    
    }

**先启动两个消费者，在启动生产者就能获得消息。**

![这里写图片描述][SouthEast 18]

![这里写图片描述][SouthEast 19]

![这里写图片描述][SouthEast 20]

![这里写图片描述][SouthEast 21]

附源码下载（打包好了，导入就可以使用）：[ActiveMQ环境搭建及实例详解的源码][ActiveMQ]

开发环境：  
window：64位  
JDK：1.8

[http_pan.baidu.com_s_1dF71zGH]: http://pan.baidu.com/s/1dF71zGH
[SouthEast]: /images/20220605/963d34efabc3423f81770697554ea1f1.png
[SouthEast 1]: /images/20220605/b7a709b98ee54435ac40c490c0595159.png
[http_localhost_8161_admin_admin_admin]: http://localhost:8161/admin%E8%A6%81%E6%B1%82%E8%BE%93%E5%85%A5%E7%94%A8%E6%88%B7%E5%90%8D%E5%AF%86%E7%A0%81%EF%BC%8C%E9%BB%98%E8%AE%A4%E7%94%A8%E6%88%B7%E5%90%8D%E5%AF%86%E7%A0%81%E4%B8%BAadmin%E3%80%81admin%EF%BC%8C%E8%BF%99%E4%B8%AA%E7%94%A8%E6%88%B7%E5%90%8D%E5%AF%86%E7%A0%81%E6%98%AF%E5%9C%A8
[SouthEast 2]: /images/20220605/663ab529e28440a3b247fbe72daab8d8.png
[SouthEast 3]: /images/20220605/768ae87d7a1940c6980cf9ea7e48b629.png
[SouthEast 4]: /images/20220605/14c19bc6cfd04c859f3fd31ce14d36ed.png
[SouthEast 5]: /images/20220605/7844e42ee05e41d1b520c14c95e4f16a.png
[SouthEast 6]: /images/20220605/3f7543abde7644c0b94a18aaedfe5223.png
[SouthEast 7]: /images/20220605/2beccf51ece54d28b4372b8402419dd3.png
[SouthEast 8]: /images/20220605/9446195afdb84b70953711f2d68075c7.png
[SouthEast 9]: /images/20220605/64ee4255759b4649bf9509b7dcf2fe72.png
[SouthEast 10]: /images/20220605/70aed17e85d84e3a967cf9e9d398da96.png
[SouthEast 11]: /images/20220605/c8b08508823744dcbcdf36ffefc5c62e.png
[SouthEast 12]: /images/20220605/abc5cf1b8df041acbee52a36130106e2.png
[SouthEast 13]: /images/20220605/3ca231647fbd4c078724214f8c863f8e.png
[SouthEast 14]: /images/20220605/66e9c70e03284964859e551277689c51.png
[SouthEast 15]: /images/20220605/4a0433a939d143a6b49be08764e02664.png
[SouthEast 16]: /images/20220605/8374670f92f04166a510523bf803d99b.png
[SouthEast 17]: /images/20220605/f612bae6aa0e4d81b0fd0ba048fa0b8b.png
[SouthEast 18]: /images/20220605/da86997562ac4e8ea9353254aa4baae6.png
[SouthEast 19]: /images/20220605/e4b610649fdd49a4a918295afc66c542.png
[SouthEast 20]: /images/20220605/f581e7afb4144a218119170510c61632.png
[SouthEast 21]: /images/20220605/4c73bb3a99f947f3b403974c1255bc00.png
[ActiveMQ]: http://download.csdn.net/download/newbie_907486852/10122246