ActiveMQ的安裝與使用。

1、什么是ActiveMQ

ActiveMQ 是Apache出品，最流行的，能力強勁的開源消息總線。ActiveMQ 是一個完全支持JMS1.1和J2EE 1.4規范的 JMS Provider實現,盡管JMS規范出台已經是很久的事情了,但是JMS在當今的J2EE應用中間仍然扮演着特殊的地位。
主要特點：
　　1）. 多種語言和協議編寫客戶端。語言: Java, C, C++, C#, Ruby, Perl, Python, PHP。應用協議: OpenWire,Stomp REST,WS Notification,XMPP,AMQP
　　2）. 完全支持JMS1.1和J2EE 1.4規范 (持久化,XA消息,事務)。
　　3.） 對Spring的支持,ActiveMQ可以很容易內嵌到使用Spring的系統里面去,而且也支持Spring2.0的特性。
　　4.） 通過了常見J2EE服務器(如 Geronimo,JBoss 4, GlassFish,WebLogic)的測試,其中通過JCA 1.5 resource adaptors的配置,可以讓ActiveMQ可以自動的部署到任何兼容J2EE 1.4 商業服務器上。
　　5）. 支持多種傳送協議:in-VM,TCP,SSL,NIO,UDP,JGroups,JXTA。
　　6）. 支持通過JDBC和journal提供高速的消息持久化。
　　7）. 從設計上保證了高性能的集群,客戶端-服務器,點對點。
　　8）. 支持Ajax。
　　9）. 支持與Axis的整合。
　　10）. 可以很容易得調用內嵌JMS provider,進行測試。

2、JMS介紹：

1）、JMS的全稱是Java Message Service，即Java消息服務。用於在兩個應用程序之間，或分布式系統中發送消息，進行異步通信。

2）、它主要用於在生產者和消費者之間進行消息傳遞，生產者負責產生消息，而消費者負責接收消息。把它應用到實際的業務需求中的話我們可以在特定的時候利用生產者生成一消息，並進行發送，對應的消費者在接收到對應的消息后去完成對應的業務邏輯。

3、ActiveMQ的兩種消息形式。

　　1）、對於消息的傳遞有兩種類型。
　　　　a）、一種是點對點的，即一個生產者和一個消費者一一對應。
　　　　b）、另一種是發布/訂閱模式，即一個生產者產生消息並進行發送后，可以由多個消費者進行接收。

　　2）、JMS定義了五種不同的消息正文格式，以及調用的消息類型，允許你發送並接收以一些不同形式的數據，提供現有消息格式的一些級別的兼容性。
　　　　a)、 StreamMessage -- Java原始值的數據流。
　　　　b)、 MapMessage--一套名稱-值對。
　　　　c)、 TextMessage--一個字符串對象。
　　　　d)、 ObjectMessage--一個序列化的 Java對象。
　　　　e)、 BytesMessage--一個字節的數據流。

4、ActiveMQ的安裝。官方網址：http://activemq.apache.org/

由於ActiveMQ是java開發的，所以需要先安裝jdk（注意：安裝jdk，需要jdk1.7以上版本）的哦。這里使用的是apache-activemq-5.12.0-bin.tar.gz版本的。

開始進行解壓縮操作。

[root@localhost package]# ls
apache-activemq-5.12.0-bin.tar.gz  apache-activemq-5.12.0-bin.zip  apache-tomcat-7.0.47.tar.gz  IK Analyzer 2012FF_hf1  IK Analyzer 2012FF_hf1.rar  jdk-7u55-linux-i586.tar.gz  solr-4.10.3.tgz.tgz  zookeeper-3.4.6.tar.gz
[root@localhost package]# tar -zxvf apache-activemq-5.12.0-bin.tar.gz -C /home/hadoop/soft/

解壓縮完以后進入bin目錄。開始進行啟動操作。

啟動：[root@localhost bin]# ./activemq start

停止：[root@localhost bin]# ./activemq stop

查看狀態：[root@localhost bin]# ./activemq status

[root@localhost soft]# cd apache-activemq-5.12.0/
[root@localhost apache-activemq-5.12.0]# ls
activemq-all-5.12.0.jar  bin  conf  data  docs  examples  lib  LICENSE  NOTICE  README.txt  webapps  webapps-demo
[root@localhost apache-activemq-5.12.0]# ll
total 9384
-rwxr-xr-x. 1 root root 9524668 Aug 10  2015 activemq-all-5.12.0.jar
drwxr-xr-x. 5 root root    4096 Sep 15 00:39 bin
drwxr-xr-x. 2 root root    4096 Sep 15 00:39 conf
drwxr-xr-x. 2 root root    4096 Sep 15 00:39 data
drwxr-xr-x. 2 root root    4096 Sep 15 00:39 docs
drwxr-xr-x. 8 root root    4096 Sep 15 00:39 examples
drwxr-xr-x. 6 root root    4096 Sep 15 00:39 lib
-rw-r--r--. 1 root root   40580 Aug 10  2015 LICENSE
-rw-r--r--. 1 root root    3334 Aug 10  2015 NOTICE
-rw-r--r--. 1 root root    2610 Aug 10  2015 README.txt
drwxr-xr-x. 7 root root    4096 Sep 15 00:39 webapps
drwxr-xr-x. 3 root root    4096 Sep 15 00:39 webapps-demo
[root@localhost apache-activemq-5.12.0]# cd bin/
[root@localhost bin]# ls
activemq  activemq-diag  activemq.jar  env  linux-x86-32  linux-x86-64  macosx  wrapper.jar
[root@localhost bin]# ./activemq start
INFO: Loading '/home/hadoop/soft/apache-activemq-5.12.0//bin/env'
INFO: Using java '/home/hadoop/soft/jdk1.7.0_55/bin/java'
INFO: Starting - inspect logfiles specified in logging.properties and log4j.properties to get details
INFO: pidfile created : '/home/hadoop/soft/apache-activemq-5.12.0//data/activemq.pid' (pid '9318')
[root@localhost bin]# ./activemq status
INFO: Loading '/home/hadoop/soft/apache-activemq-5.12.0//bin/env'
INFO: Using java '/home/hadoop/soft/jdk1.7.0_55/bin/java'
ActiveMQ is running (pid '9318')
[root@localhost bin]# 

然后你可以訪問后台管理界面，賬號和密碼默認都是admin的。訪問地址：http://192.168.110.142:8161/admin

Home是當前的歡迎頁，Queues是點到點形式，Topics是發布訂閱模式，Subscribers話題消息的發布與訂閱，Connections客戶端鏈接，Network當前網絡的鏈接狀態，Scheduled計划任務，Send可以測試發送消息。

5、ActiveMQ的使用方法，JMS消息發送模式。

注意：
1）、在點對點或隊列模型下，一個生產者向一個特定的隊列發布消息，一個消費者從該隊列中讀取消息。這里，生產者知道消費者的隊列，並直接將消息發送到消費者的隊列。這種模式被概括為：只有一個消費者將獲得消息。生產者不需要在接收者消費該消息期間處於運行狀態，接收者也同樣不需要在消息發送時處於運行狀態。每一個成功處理的消息都由接收者簽收。
2）、發布者/訂閱者模型支持向一個特定的消息主題發布消息。0或多個訂閱者可能對接收來自特定消息主題的消息感興趣。在這種模型下，發布者和訂閱者彼此不知道對方。這種模式好比是匿名公告板。這種模式被概括為：多個消費者可以獲得消息，在發布者和訂閱者之間存在時間依賴性。發布者需要建立一個訂閱（subscription），以便客戶能夠購訂閱。訂閱者必須保持持續的活動狀態以接收消息，除非訂閱者建立了持久的訂閱。在那種情況下，在訂閱者未連接時發布的消息將在訂閱者重新連接時重新發布。

6、JMS應用程序接口。

1）、ConnectionFactory 接口（連接工廠）
    用戶用來創建到JMS提供者的連接的被管對象。JMS客戶通過可移植的接口訪問連接，這樣當下層的實現改變時，代碼不需要進行修改。 管理員在JNDI名字空間中配置連接工廠，這樣，JMS客戶才能夠查找到它們。根據消息類型的不同，用戶將使用隊列連接工廠，或者主題連接工廠。
2）、Connection 接口（連接）
    連接代表了應用程序和消息服務器之間的通信鏈路。在獲得了連接工廠后，就可以創建一個與JMS提供者的連接。根據不同的連接類型，連接允許用戶創建會話，以發送和接收隊列和主題到目標。
3）、Destination 接口（目標）
    目標是一個包裝了消息目標標識符的被管對象，消息目標是指消息發布和接收的地點，或者是隊列，或者是主題。JMS管理員創建這些對象，然后用戶通過JNDI發現它們。和連接工廠一樣，管理員可以創建兩種類型的目標，點對點模型的隊列，以及發布者／訂閱者模型的主題。
4）、MessageConsumer 接口（消息消費者）
    由會話創建的對象，用於接收發送到目標的消息。消費者可以同步地（阻塞模式），或異步（非阻塞）接收隊列和主題類型的消息。
5）、MessageProducer 接口（消息生產者）
    由會話創建的對象，用於發送消息到目標。用戶可以創建某個目標的發送者，也可以創建一個通用的發送者，在發送消息時指定目標。
6）、Message 接口（消息）
    是在消費者和生產者之間傳送的對象，也就是說從一個應用程序創送到另一個應用程序。一個消息有三個主要部分：
        消息頭（必須）：包含用於識別和為消息尋找路由的操作設置。
        一組消息屬性（可選）：包含額外的屬性，支持其他提供者和用戶的兼容。可以創建定制的字段和過濾器（消息選擇器）。
        一個消息體（可選）：允許用戶創建五種類型的消息（文本消息，映射消息，字節消息，流消息和對象消息）。
        消息接口非常靈活，並提供了許多方式來定制消息的內容。
7）、Session 接口（會話）
    表示一個單線程的上下文，用於發送和接收消息。由於會話是單線程的，所以消息是連續的，就是說消息是按照發送的順序一個一個接收的。會話的好處是它支持事務。如果用戶選擇了事務支持，會話上下文將保存一組消息，直到事務被提交才發送這些消息。在提交事務之前，用戶可以使用回滾操作取消這些消息。一個會話允許用戶創建消息生產者來發送消息，創建消息消費者來接收消息。

7、如何使用java操作activeMQ呢，把ActiveMQ依賴的jar包添加到工程中。

使用maven工程，則添加jar包的依賴：

<dependency>
    <groupId>org.apache.activemq</groupId>
    <artifactId>activemq-all</artifactId>
    <version>5.11.2</version>
</dependency>

然后你就可以愉快得開發了。是不是很開森呢。

8、ActiveMQ點對點模式（point-to-point）。

ActiveMq的點對點生產者。

package com.taotao.activemq;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;

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

/**
 * 
 * @ClassName: ActiveMqMain.java
 * @author: biehl
 * @since: 2019年9月15日 下午4:44:57
 * @Copyright: ©2019 biehl 版權所有
 * @version: 0.0.1
 * @Description:
 */
public class ActiveMqMain {

    // activeMq得點對點生產者
    @Test
    public void queueProducer() throws JMSException {
        // 1、創建一個連接工廠對象ConnectionFactory對象。需要指定mq服務得ip以及端口號61616。
        String brokerURL = "tcp://192.168.110.142:61616";
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(brokerURL);
        // 2、使用ConnectionFactory創建一個連接Connection對象。
        Connection connection = connectionFactory.createConnection();
        // 3、開啟連接。調用Connection對象得start方法。
        connection.start();
        // 4、使用Connection對象創建一個Session對象。
        // 參數一是否開啟事務，一般不開啟事務，保證數據得最終一致性，可以使用消息隊列實現數據最終一致性。如果第一個參數為true，第二個參數自動忽略
        // 參數二是消息得應答模式。兩種模式，自動應答和手動應答。一般使用自動應答。
        boolean transacted = false;// 不開啟事務
        int acknowledgeMode = Session.AUTO_ACKNOWLEDGE;// 1
        Session session = connection.createSession(transacted, acknowledgeMode);
        // 5、使用Session對象創建一個Destination對象。兩種形式queue、topic。現在應該使用queue。
        String queueName = "queue1";// 當前消息隊列得名稱
        Queue queue = session.createQueue(queueName);
        // 6、使用Session對象創建一個Producer對象。
        // interface Queue extends Destination。destination是一個接口。
        MessageProducer producer = session.createProducer(queue);
        // 7、創建一個TextMessage對象。
        // 創建TextMessage方式一
        // TextMessage textMessage = new ActiveMQTextMessage();
        // textMessage.setText("hello activeMq......");
        // 方式二
        TextMessage textMessage = session.createTextMessage("hello activeMq......");
        // 8、發送消息。
        producer.send(textMessage);
        // 9、關閉資源。
        producer.close();// 關閉producer
        session.close();// 關閉session
        connection.close();// 關閉connection
    }

}

ActiveMQ的點對點消息生產成功以后，可以在ActiveMQ提供的web界面可以看到一些信息。

activeMq的點對點消費者。

package com.taotao.activemq;

import java.io.IOException;

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.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;

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

/**
 * 
 * @ClassName: ActiveMqMain.java
 * @author: biehl
 * @since: 2019年9月15日 下午4:44:57
 * @Copyright: ©2019 biehl 版權所有
 * @version: 0.0.1
 * @Description:
 */
public class ActiveMqMain {

    // activeMq的點對點生產者
    @Test
    public void queueProducer() throws JMSException {
        // 1、創建一個連接工廠對象ConnectionFactory對象。需要指定mq服務得ip以及端口號61616。
        String brokerURL = "tcp://192.168.110.142:61616";
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(brokerURL);
        // 2、使用ConnectionFactory創建一個連接Connection對象。
        Connection connection = connectionFactory.createConnection();
        // 3、開啟連接。調用Connection對象得start方法。
        connection.start();
        // 4、使用Connection對象創建一個Session對象。
        // 參數一是否開啟事務，一般不開啟事務，保證數據得最終一致性，可以使用消息隊列實現數據最終一致性。如果第一個參數為true，第二個參數自動忽略
        // 參數二是消息得應答模式。兩種模式，自動應答和手動應答。一般使用自動應答。
        boolean transacted = false;// 不開啟事務
        int acknowledgeMode = Session.AUTO_ACKNOWLEDGE;// 1
        Session session = connection.createSession(transacted, acknowledgeMode);
        // 5、使用Session對象創建一個Destination對象。兩種形式queue、topic。現在應該使用queue。
        String queueName = "queue1";// 當前消息隊列得名稱
        Queue queue = session.createQueue(queueName);
        // 6、使用Session對象創建一個Producer對象。
        // interface Queue extends Destination。destination是一個接口。
        MessageProducer producer = session.createProducer(queue);
        // 7、創建一個TextMessage對象。
        // 創建TextMessage方式一
        // TextMessage textMessage = new ActiveMQTextMessage();
        // textMessage.setText("hello activeMq......");
        // 方式二
        TextMessage textMessage = session.createTextMessage("hello activeMq......");
        // 8、發送消息。
        producer.send(textMessage);
        // 9、關閉資源。
        producer.close();// 關閉producer
        session.close();// 關閉session
        connection.close();// 關閉connection
    }

    // activeMq的點對點消費者
    @Test
    public void queueConsumer() throws JMSException {
        // 1、創建一個連接工廠ConnectionFactory 對象
        String brokerURL = "tcp://192.168.110.142:61616";
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(brokerURL);
        // 2、使用連接工廠對象創建一個連接
        Connection connection = connectionFactory.createConnection();
        // 3、開啟連接
        connection.start();
        // 4、使用連接對象創建一個Session對象
        boolean transacted = false;// 關閉事務
        int acknowledgeMode = Session.AUTO_ACKNOWLEDGE;// 自動響應
        Session session = connection.createSession(transacted, acknowledgeMode);
        // 5、使用Session創建一個Destination，Destination應該和消息的發送端一致的。
        String queueName = "queue1";
        Queue queue = session.createQueue(queueName);
        // 6、使用Session創建一個Consumer對象。
        MessageConsumer consumer = session.createConsumer(queue);
        // 7、向Consumer對象中設置一個MessageListener對象，用來接受消息。
        // 匿名內部類，new 接口，后面加上{}，相當於實現了這個接口的實現類。然后創建這個實現類的對象listener。
        MessageListener listener = new MessageListener() {

            @Override
            public void onMessage(Message message) {
                // 接受事件的。當消息到達就可以在這里接受到消息了的。
                // 8、取出消息的內容。
                if (message instanceof TextMessage) {
                    TextMessage textMessage = (TextMessage) message;
                    // 9、打印消息內容。
                    try {
                        String text = textMessage.getText();
                        System.out.println(text);
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            }
        };
        consumer.setMessageListener(listener);

        // 關閉資源以前，系統等待，等待接受消息。
        /*while (true) {
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }*/
        
        // 等待鍵盤輸入。才回接着向下執行的。
        try {
            System.in.read();
        } catch (IOException e) {
            e.printStackTrace();
        }        
        
        
        // 10、關閉資源。
        consumer.close();// 關閉consumer
        session.close();// 關閉session
        connection.close();// 關閉connection
    }

}

執行了activeMq的點對點消費者。可以在界面看到變化。可以看到有一個消費者，然后生產了7條消息，7條消息進隊和7條消息出隊。

 9、ActiveMQ發布訂閱模式（publish/subscribe）。

消費者有兩種消費方法（這里使用異步消費）:
　　a、同步消費。通過調用消費者的receive方法從目的地中顯式提取消息。receive方法可以一直阻塞到消息到達。

　　b、異步消費。客戶可以為消費者注冊一個消息監聽器，以定義在消息到達時所采取的動作。

　　　　實現MessageListener接口，在MessageListener（）方法中實現消息的處理邏輯。

package com.taotao.activemq;

import java.io.IOException;

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.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;

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

/**
 * Active的發布訂閱模式
 * 
 * @ClassName: ActiveMqTopics.java
 * @author: biehl
 * @since: 2019年9月19日 上午10:51:14
 * @Copyright: ©2019 biehl 版權所有
 * @version: 0.0.1
 * @Description:
 */
public class ActiveMqTopics {

    // 發布訂閱模式，生產者。topic生產者生產消息默認不持久化客戶端的。
    @Test
    public void topicProducer() {
        try {
            // 1、創建一個連接工廠對象。需要指定mq服務的ip地址以及端口號61616
            String brikerURL = "tcp://192.168.110.142:61616";
            // 創建ConnectionFactory接口對象，實現類ActiveMQConnectionFactory
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(brikerURL);

            // 2、創建Connection連接
            Connection connection = connectionFactory.createConnection();

            // 3、開啟連接，調用Connection的start方法。
            connection.start();

            // 4、創建Session，使用Connection對象創建一個session
            // 參數一是否開啟事務，一般不開啟事務，保證數據得最終一致性，可以使用消息隊列實現數據最終一致性。如果第一個參數為true，第二個參數自動忽略
            boolean transacted = false;
            // 參數二是消息得應答模式。兩種模式，自動應答和手動應答。一般使用自動應答。
            int acknowledgeMode = Session.AUTO_ACKNOWLEDGE;
            Session session = connection.createSession(transacted, acknowledgeMode);

            // 5、創建Destination，應該使用topic，區別於點對點的queue
            String topicName = "topic01";
            Topic topic = session.createTopic(topicName);

            // 6、創建一個Producer對象
            // interface Topic extends Destination.
            // Destination是一個接口，Topic接口繼承Destination這個接口。
            MessageProducer producer = session.createProducer(topic);

            // 7、創建一個TextMessage對象
            String message = null;
            TextMessage textMessage = null;
            for (int i = 0; i < 100; i++) {
                message = i + " ActiveMQ topics......";
                textMessage = session.createTextMessage(message);

                // 8、發送消息
                producer.send(textMessage);
            }

            // 9、關閉資源
            producer.close();// 關閉producer
            session.close();// 關閉session
            connection.close();// 關閉connection
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }

    // 發布訂閱模式，消費者必須一直等待生產者生產的消息，因為發布訂閱模式不持久化。
    @Test
    public void topicConsumer() {
        try {
            // 1、創建一個連接工廠對象
            String brokerURL = "tcp://192.168.110.142:61616";
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(brokerURL);

            // 2、使用連接工廠對象創建一個連接
            Connection connection = connectionFactory.createConnection();

            // 3、開啟連接
            connection.start();

            // 4、使用連接對象創建一個Session對象
            // 參數一是否開啟事務，一般不開啟事務，保證數據得最終一致性，可以使用消息隊列實現數據最終一致性。如果第一個參數為true，第二個參數自動忽略
            boolean transacted = false;
            // 參數二是消息得應答模式。兩種模式，自動應答和手動應答。一般使用自動應答。
            int acknowledgeMode = Session.AUTO_ACKNOWLEDGE;
            Session session = connection.createSession(transacted, acknowledgeMode);

            // 5、使用session創建destination，注意，destination應該和消息的發送端一致的。
            String topicName = "topic01";
            Topic topic = session.createTopic(topicName);

            // 6、使用session創建一個consumer對象
            MessageConsumer consumer = session.createConsumer(topic);

            // 7、向Consumer對象中設置一個MessageListener對象，用來接受消息。
            // 匿名內部類，new 接口，后面加上{}，相當於實現了這個接口的實現類。然后創建這個實現類的對象listener。
            MessageListener listener = new MessageListener() {
                // 接受事件的。當消息到達就可以在這里接受到消息了的。
                // 8、取出消息的內容。
                @Override
                public void onMessage(Message message) {
                    if (message instanceof TextMessage) {
                        TextMessage textMessage = (TextMessage) message;
                        // 9、打印消息內容。
                        try {
                            String text = textMessage.getText();
                            System.out.println(text);
                        } catch (JMSException e) {
                            e.printStackTrace();
                        }
                    }
                }
            };
            consumer.setMessageListener(listener);

            // 啟動三次，模擬是三個消費者
            System.out.println("消費者1.......");
            // System.out.println("消費者2.......");
            // System.out.println("消費者3.......");

            // 等待鍵盤輸入。才回接着向下執行的。
            try {
                System.in.read();
            } catch (IOException e) {
                e.printStackTrace();
            }

            // 9、關閉資源
            consumer.close();// 關閉producer
            session.close();// 關閉session
            connection.close();// 關閉connection
        } catch (JMSException e) {
            e.printStackTrace();
        }

    }

}

執行了activeMq的發布訂閱模式。可以在界面看到變化。可以看到有三個消費者，然后生產了201條消息，201條消息進隊和603條消息出隊。

10、ActiveMQ與Spring整合如下所示:

在pom.xml配置文件中引入自己的依賴的jar包。

<dependency>
    <groupId>org.springframework</groupId>
    <artifactId>spring-jms</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework</groupId>
    <artifactId>spring-context-support</artifactId>
</dependency>

在配置文件applicationContext-activemq.xml里面配置ConnectionFactory。如下所示：

<?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:jms="http://www.springframework.org/schema/jms"
    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.0.xsd
    http://www.springframework.org/schema/context 
    http://www.springframework.org/schema/context/spring-context-4.0.xsd
    http://www.springframework.org/schema/aop 
    http://www.springframework.org/schema/aop/spring-aop-4.0.xsd 
    http://www.springframework.org/schema/tx 
    http://www.springframework.org/schema/tx/spring-tx-4.0.xsd
    http://www.springframework.org/schema/jms 
    http://www.springframework.org/schema/jms/spring-jms-4.0.xsd
    http://www.springframework.org/schema/util 
    http://www.springframework.org/schema/util/spring-util-4.0.xsd">


    <!-- 真正可以產生Connection的ConnectionFactory，由對應的 JMS服務廠商提供 -->
    <bean id="targetConnectionFactory"
        class="org.apache.activemq.ActiveMQConnectionFactory">
        <property name="brokerURL"
            value="tcp://192.168.110.142:61616" />
    </bean>

    <!-- Spring用於管理真正的ConnectionFactory的ConnectionFactory -->
    <bean id="connectionFactory"
        class="org.springframework.jms.connection.SingleConnectionFactory">
        <!-- 目標ConnectionFactory對應真實的可以產生JMS Connection的ConnectionFactory -->
        <property name="targetConnectionFactory"
            ref="targetConnectionFactory" />
    </bean>
</beans>

開始配置生產者的spring配置。

<?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:jms="http://www.springframework.org/schema/jms"
    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.0.xsd
    http://www.springframework.org/schema/context 
    http://www.springframework.org/schema/context/spring-context-4.0.xsd
    http://www.springframework.org/schema/aop 
    http://www.springframework.org/schema/aop/spring-aop-4.0.xsd 
    http://www.springframework.org/schema/tx 
    http://www.springframework.org/schema/tx/spring-tx-4.0.xsd
    http://www.springframework.org/schema/jms 
    http://www.springframework.org/schema/jms/spring-jms-4.0.xsd
    http://www.springframework.org/schema/util 
    http://www.springframework.org/schema/util/spring-util-4.0.xsd">


    <!-- 1、真正可以產生Connection的ConnectionFactory，由對應的 JMS服務廠商提供 -->
    <bean id="targetConnectionFactory"
        class="org.apache.activemq.ActiveMQConnectionFactory">
        <property name="brokerURL"
            value="tcp://192.168.110.142:61616" />
    </bean>

    <!-- 2、Spring用於管理真正的ConnectionFactory的ConnectionFactory -->
    <bean id="connectionFactory"
        class="org.springframework.jms.connection.SingleConnectionFactory">
        <!-- 目標ConnectionFactory對應真實的可以產生JMS Connection的ConnectionFactory -->
        <!-- 給屬性targetConnectionFactory傳值 -->
        <property name="targetConnectionFactory"
            ref="targetConnectionFactory" />
    </bean>

    <!-- 3、開始配置生產者配置 -->
    <!-- 配置生產者 -->
    <!-- Spring提供的JMS工具類，它可以進行消息發送、接收等 -->
    <bean id="jmsTemplate"
        class="org.springframework.jms.core.JmsTemplate">
        <!-- 這個connectionFactory對應的是我們定義的Spring提供的那個ConnectionFactory對象 -->
        <!-- 給屬性connectionFactory傳值 -->
        <property name="connectionFactory" ref="connectionFactory" />
    </bean>

    <!-- 4、配置消息的Destination對象 -->
    <!-- 點對點模式 -->
    <!-- 這個是隊列目的地，點對點的。 -->
    <bean id="queueDestination"
        class="org.apache.activemq.command.ActiveMQQueue">
        <constructor-arg>
            <!-- 給ActiveMQQueue構造參數傳遞一個值為queue -->
            <value>queue</value>
        </constructor-arg>
    </bean>

    <!-- 發布訂閱模式 -->
    <!-- 這個是主題目的地，一對多的。 -->
    <bean id="topicDestination"
        class="org.apache.activemq.command.ActiveMQTopic">
        <!-- 給ActiveMQTopic構造參數傳遞一個值為topic -->
        <constructor-arg value="topic" />
    </bean>

</beans>

生產者測試代碼如下所示：

可以根據之前的消費者測試一下，消息的消費。

package com.taotao.activemq;

import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.junit.Test;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;

/**
 * 
 * @ClassName: SpringActiveMQ.java
 * @author: biehl
 * @since: 2019年9月19日 下午7:01:43
 * @Copyright: ©2019 biehl 版權所有
 * @version: 0.0.1
 * @Description:
 */
public class SpringActiveMQ {

    // 使用spring與activemq整合，是喲個jmsTemplate發送消息
    @Test
    public void jmsTemplateProducer() {
        // 1、初始化spring容器
        ApplicationContext applicationContext = new ClassPathXmlApplicationContext(
                "classpath:/spring/applicationContext-activemq.xml");
        // 2、從容器中獲得jmsTemplate對象。根據類型獲取到bean的對象
        JmsTemplate jmsTemplate = applicationContext.getBean(JmsTemplate.class);
        // 3、從容器中獲得Destination對象。根據名稱獲取到bean的對象
        Destination destination = (Destination) applicationContext.getBean("queueDestination");

        // 4、發送消息
        jmsTemplate.send(destination, new MessageCreator() {

            @Override
            public Message createMessage(Session session) throws JMSException {
                // 定義一個消息
                String message = "hello activeMq......";
                // 發送消息
                TextMessage textMessage = session.createTextMessage(message);
                return textMessage;
            }
        });
    }

}

效果如下所示：

 開始配置消費者的spring配置。

　　1）、注意：那么消費者是通過Spring為我們封裝的消息監聽容器MessageListenerContainer實現的，它負責接收信息，並把接收到的信息分發給真正的MessageListener進行處理。每個消費者對應每個目的地都需要有對應的MessageListenerContainer。
　　2）、對於消息監聽容器而言，除了要知道監聽哪個目的地之外，還需要知道到哪里去監聽，也就是說它還需要知道去監聽哪個JMS服務器，這是通過在配置MessageConnectionFactory的時候往里面注入一個ConnectionFactory來實現的。
　　3）、所以在配置一個MessageListenerContainer的時候有三個屬性必須指定：
　　　　a、一個是表示從哪里監聽的ConnectionFactory
　　　　b、一個是表示監聽什么的Destination；
　　　　c、一個是接收到消息以后進行消息處理的MessageListener。
　　4）、常用的MessageListenerContainer實現類是DefaultMessageListenerContainer。

<?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:jms="http://www.springframework.org/schema/jms"
    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.0.xsd
    http://www.springframework.org/schema/context 
    http://www.springframework.org/schema/context/spring-context-4.0.xsd
    http://www.springframework.org/schema/aop 
    http://www.springframework.org/schema/aop/spring-aop-4.0.xsd 
    http://www.springframework.org/schema/tx 
    http://www.springframework.org/schema/tx/spring-tx-4.0.xsd
    http://www.springframework.org/schema/jms 
    http://www.springframework.org/schema/jms/spring-jms-4.0.xsd
    http://www.springframework.org/schema/util 
    http://www.springframework.org/schema/util/spring-util-4.0.xsd">


    <!-- 1、真正可以產生Connection的ConnectionFactory，由對應的 JMS服務廠商提供 -->
    <bean id="targetConnectionFactory"
        class="org.apache.activemq.ActiveMQConnectionFactory">
        <property name="brokerURL"
            value="tcp://192.168.110.142:61616" />
    </bean>

    <!-- 2、Spring用於管理真正的ConnectionFactory的ConnectionFactory -->
    <bean id="connectionFactory"
        class="org.springframework.jms.connection.SingleConnectionFactory">
        <!-- 目標ConnectionFactory對應真實的可以產生JMS Connection的ConnectionFactory -->
        <!-- 給屬性targetConnectionFactory傳值 -->
        <property name="targetConnectionFactory"
            ref="targetConnectionFactory" />
    </bean>

    <!-- 3、配置消息的Destination對象。接受消息的目的地。 -->
    <!-- 點對點模式 -->
    <!-- 這個是隊列目的地，點對點的。 -->
    <bean id="queueDestination"
        class="org.apache.activemq.command.ActiveMQQueue">
        <constructor-arg>
            <!-- 給ActiveMQQueue構造參數傳遞一個值為queue -->
            <value>queue</value>
        </constructor-arg>
    </bean>

    <!-- 發布訂閱模式 -->
    <!-- 這個是主題目的地，一對多的。 -->
    <bean id="topicDestination"
        class="org.apache.activemq.command.ActiveMQTopic">
        <!-- 給ActiveMQTopic構造參數傳遞一個值為topic -->
        <constructor-arg value="topic" />
    </bean>

    <!-- 4、配置消息接收者 -->
    <!-- 配置一個監聽器 -->
    <bean id="activeMqMessageListener"
        class="com.taotao.search.listener.ActiveMqMessageListener" />

    <!-- 配置監聽容器 -->
    <bean
        class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <!-- 屬性設置 -->
        <!-- 一個是表示從哪里監聽的ConnectionFactory -->
        <property name="connectionFactory" ref="connectionFactory" />
        <!-- 一個是表示監聽什么的Destination -->
        <property name="destination" ref="queueDestination" />
        <!-- 一個是接收到消息以后進行消息處理的MessageListener -->
        <property name="messageListener" ref="activeMqMessageListener" />
    </bean>


</beans>

然后可以寫消息監聽器，用來監聽生產者生產的消息，以便實現自己的業務邏輯。

package com.taotao.search.listener;

import java.text.SimpleDateFormat;
import java.util.Date;

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

/**
 * 接受ActiveMQ發送的消息.
 * 
 * @ClassName: ActiveMqMessageListener.java
 * @author: biehl
 * @since: 2019年9月19日 下午7:55:24
 * @Copyright: ©2019 biehl 版權所有
 * @version: 0.0.1
 * @Description:
 */
public class ActiveMqMessageListener implements MessageListener {

    @Override
    public void onMessage(Message message) {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss");
        System.out.println("監聽生產者生產的消息，消費者進行消息消費.......");
        // 消息到了onMessage就接受到了消息
        if (message instanceof TextMessage) {
            TextMessage textMessage = (TextMessage) message;
            try {
                String text = textMessage.getText();
                System.out.println(sdf.format(new Date()) + " : " + text);
            } catch (JMSException e) {
                e.printStackTrace();
            }
        }
    }

}

由於這里只是簡單的測試，如果是正式項目的話，直接加載這個配置文件，然后就可以進行消息的監聽消費，我這里只是加載一下這個配置文件即可。

package com.taotao.search.service;

import java.io.IOException;

import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

/**
 * 
 * @ClassName: ActiveMqConsumer.java
 * @author: biehl
 * @since: 2019年9月19日 下午8:10:55
 * @Copyright: ©2019 biehl 版權所有
 * @version: 0.0.1
 * @Description:
 */
public class ActiveMqConsumer {

    // 啟動spring容器。就可以實現監聽生產者發送消息，消費者消費小的目的地。
    public static void main(String[] args) {
        // 初始化spring容器
        ApplicationContext applicationContext = new ClassPathXmlApplicationContext(
                "classpath:/spring/applicationContext-activemq.xml");
        System.out.println("spring容器加載完畢，開始監聽生產者生產的消息.......");
        try {
            System.in.read();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

實現效果如下所示：

控制台打印如下所示，只要你生產消息，這里就可以進行消息的消費。

 

 

待續......