文章目錄

• 1、同步索引庫問題分析
• 2、ActiveM的介紹
  • 2.1、什麼是ActiveMQ
  • 2.2、ActiveMQ的訊息形式
• 3、ActiveMQ的安裝
  • 3.1、安裝環境
  • 3.2、安裝步驟
• 4、ActiveMQ的使用方法
  • 4.1、Queue（佇列）
  • 4.2、Topic（話題）
• 5、ActiveMQ整合spring
  • 5.1、使用方法
  • 5.2、程式碼測試（Queue方式）
• 6、ActiveMQ整合到專案中
  • 6.1、Producer
  • 6.2、Consumer

 

---

課程計劃

• 1、同步索引庫問題分析
• 2、什麼是MQ
• 3、ActiveMQ的安裝
• 4、ActiveMQ的使用方法
• 5、ActiveMQ整合spring
• 6、使用訊息佇列實現新增商品後同步索引庫

回到頂部

1、同步索引庫問題分析

方案一：在taotao-manager中，新增商品的業務邏輯中，新增一個同步索引庫的業務邏輯。

  缺點： 業務邏輯耦合度非常高，業務邏輯拆分不明確。（單一職能原則）
方案二：業務邏輯在taotao-search中實現，呼叫服務在taotao-manager實現。業務邏輯分開。既可以是服務的生產者，也可以是服務的消費者。

  缺點： 服務之間的耦合度變高。服務的 啟動有先後順序。隨著呼叫的服務會越來越多， 服務之間的呼叫越來越複雜，難以管理。
方案三：使用訊息佇列。MQ是一個訊息中介軟體。

MQ就相當於一個祕書、一個運營商，如下圖所示：

存在的問題：
  1、如果MQ掛了，所有相關的服務都掛了。
  2、MQ有效能的瓶頸，儘量減少訊息的內容的大小。
注意： 技術的選型和具體的業務有關，只選擇合適的技術。
如果MQ掛了：
  1、通過日誌查詢原因。
  2、通知相關的人員修復。
  3、關鍵的業務必須保證有備用方案。 回到頂部

2、ActiveM的介紹

  MQ是一個訊息中介軟體，比如：ActiveMQ(Java)、RabbitMQ(Erlang)、Kafka(大資料)都屬於MQ，是MQ的產品。

2.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.2、ActiveMQ的訊息形式

  對於訊息的傳遞有兩種型別：
  一種是點對點的，即一個生產者和一個消費者一一對應。（簡訊）
  另一種是釋出/訂閱模式，即一個生產者產生訊息並進行傳送後，可以由多個消費者進行接收。（廣播）
  JMS定義了五種不同的訊息正文格式，以及呼叫的訊息型別，允許你傳送並接收以一些不同形式的資料，提供現有訊息格式的一些級別的相容性。
    StreamMessage --> Java原始值的資料流
    MapMessage --> 一套名稱-值對
    TextMessage --> 一個字串物件
    ObjectMessage --> 一個序列化的 Java物件
    BytesMessage --> 一個位元組的資料流

回到頂部

3、ActiveMQ的安裝

網址：http://activemq.apache.org/

下載ActiveMQ，使用的版本是5.13.0，最新的版本是5.15.8。

3.1、安裝環境

1、activemq是java語言開發的，需要安裝jdk。
2、安裝Linux系統。生產環境都是Linux系統。

3.2、安裝步驟

第一步：把 apache-activemq-5.13.0-bin.tar.gz 的壓縮包上傳到Linux系統。
第二步：解壓縮後刪除壓縮包。

[[email protected] ~]# ll
總用量 65472
-rw-r--r--. 1 root root 50385720 11月 25 12:32 apache-activemq-5.13.0-bin.tar.gz
[[email protected] ~]# tar zxf apache-activemq-5.13.0-bin.tar.gz
[[email protected] ~]# ll
總用量 52360
drwxr-xr-x. 10 root root      193 11月 30 2015 apache-activemq-5.13.0
-rw-r--r--.  1 root root 53613208 1月   7 2016 apache-activemq-5.13.0-bin.tar.gz
[[email protected] ~]# rm -rf apache-activemq-5.13.0-bin.tar.gz 

第三步：啟動。
使用bin目錄下的activemq命令

啟動：
[[email protected] bin]# ./activemq start
關閉：
[[email protected] bin]# ./activemq stop
檢視狀態：
[[email protected] bin]# ./activemq status
檢視執行埠號：
[[email protected] bin]# ps aux | grep activemq

第四步：本博主的是CentOS7，防火牆使用的是firewalld，我們使用修改配置檔案的方式來新增用到的埠（修改後需要重啟firewalld服務）

[[email protected] bin]# cd /etc/firewalld/zones/
[[email protected] zones]# pwd
/etc/firewalld/zones
[[email protected] zones]# vim public.xml

編輯public.xml檔案，新增程式碼：<port protocol="tcp" port="8161"/>
儲存退出後，然後我們需要重啟firewalld服務：

[[email protected] zones]# service firewalld restart
Redirecting to /bin/systemctl restart firewalld.service
[[email protected] zones]#

第五步：進入activemq管理後臺。

訪問地址：http://192.168.25.168:8161/admin
使用者名稱：admin
密碼：admin

如下圖所示：

解決ActiveMQ訪問後臺時出現503錯誤（點選Queues按鈕時）：

1、檢視機器名
[[email protected] bin]# vim /etc/sysconfig/network
NETWORKING=yes
HOSTNAME=taotao.com

2、修改機器名和ip的對映關係，即修改hosts檔案
[[email protected] bin]# vim /etc/hosts
127.0.0.1   localhost localhost.localdomain localhost4 localhost4.localdomain4 taotao
::1         localhost localhost.localdomain localhost6 localhost6.localdomain6

3、重啟ActiveMQ服務

回到頂部

4、ActiveMQ的使用方法

點對點：

釋出/訂閱：

4.1、Queue（佇列）

  ActiveMQ的Queue方式預設在ActiveMQ的服務端是持久化快取的。

4.1.1、Producer

本例中，我們使用taotao-manager-service工程作為生產者。
生產者：生產訊息，傳送端。
把jar包新增到工程中。使用5.11.2版本的jar包。版本號我們在taotao-parent中的pom.xml中配置。

第一步：建立ConnectionFactory物件，需要指定服務端ip及埠號61616。注意：8161是ActiveMQ的web服務的埠。
第二步：使用ConnectionFactory物件建立一個Connection物件。
第三步：開啟連線，呼叫Connection物件的start()方法。
第四步：使用Connection物件建立一個Session物件。
第五步：使用Session物件建立一個Destination物件（topic、queue），此處建立一個Queue物件。
第六步：使用Session物件建立一個Producer物件。
第七步：建立一個Message物件，建立一個TextMessage物件。
第八步：使用Producer物件傳送訊息。
第九步：關閉資源。
我們在taotao-manager-service工程中新建測試類和測試方法。
測試程式碼如下：

    /**
     * ActiveMQ的Queue方式的Producer
     * @throws Exception
     */
    @Test
    public void queueProducerTest() throws Exception {
        // 第一步：建立ConnectionFactory物件，需要指定服務端ip及埠號61616。注意：8161是ActiveMQ的web服務的埠。
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.168:61616"); // 注意：虛擬機器上的安裝ActiveMQ服務要新增該埠
        // 第二步：使用ConnectionFactory物件建立一個Connection物件。
        Connection connection = connectionFactory.createConnection();
        // 第三步：開啟連線，呼叫Connection物件的start方法。
        connection.start();
        // 第四步：使用Connection物件建立一個Session物件。
        // 第一個引數：是否開啟分散式事務。true：開啟分散式事務，當第一個引數為true時，第二個引數忽略。
        // 一般我們不使用分散式事務，因為分散式事務複雜，效能不好。網際網路專案要求快速響應給使用者，其餘的事情交給訊息佇列向各個服務發訊息讓他們去做。這樣保證了資料的最終一致。
        // 第二個引數：當第一個引數為false時，第二個引數才有意義，表示訊息的應答模式。1、自動應答  2、手動應答。一般是自動應答。
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 第五步：使用Session物件建立一個Destination物件（topic、queue），此處建立一個Queue物件。引數表示佇列的名稱。
        Queue queue = session.createQueue("test-queue");
        // 第六步：使用Session物件建立一個Producer物件。
        MessageProducer producer = session.createProducer(queue);
        // 第七步：建立一個Message物件，建立一個TextMessage物件。
        // 方式一：
        // TextMessage message = new ActiveMQTextMessage();
        // message.setText("hello ActiveMQ, this is my first test.");
        // 方式二：
        TextMessage textMessage = session.createTextMessage("hello ActiveMQ, this is my first test.");
        // 第八步：使用Producer物件傳送訊息。
        producer.send(textMessage);
        // 第九步：關閉資源。
        producer.close();
        session.close();
        connection.close();
    }

進入activemq管理後臺檢視新增的queues佇列，如下圖：

4.1.2、Consumer

本例中，我們使用taotao-search-service工程作為消費者。
消費者：接收訊息。
第一步：建立一個ConnectionFactory物件。
第二步：從ConnectionFactory物件中獲得一個Connection物件。
第三步：開啟連線。呼叫Connection物件的start()方法。
第四步：使用Connection物件建立一個Session物件。
第五步：使用Session物件建立一個Destination物件。和傳送端保持一致queue，並且佇列的名稱一致。
第六步：使用Session物件建立一個Consumer物件。
第七步：接收訊息。
第八步：列印訊息。
第九步：關閉資源。
測試程式碼如下：

    /**
     * ActiveMQ的Queue方式的Consumer
     * @throws Exception
     */
    @Test
    public void queueConsumerTest() throws Exception {
        // 第一步：建立一個ConnectionFactory物件。
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.168:61616");
        // 第二步：從ConnectionFactory物件中獲得一個Connection物件。
        Connection connection = connectionFactory.createConnection();
        // 第三步：開啟連線。呼叫Connection物件的start方法。
        connection.start();
        // 第四步：使用Connection物件建立一個Session物件。
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 第五步：使用Session物件建立一個Destination物件。和傳送端保持一致queue，並且佇列的名稱一致。
        Queue queue = session.createQueue("test-queue");
        // 第六步：使用Session物件建立一個Consumer物件。
        MessageConsumer consumer = session.createConsumer(queue);
        // 第七步：接收訊息。
        consumer.setMessageListener(new MessageListener() {

            @Override
            public void onMessage(Message message) {
                if (message instanceof TextMessage) {
                    TextMessage textMessage = (TextMessage) message;
                    try {
                        String text = null;
                        // 取訊息的內容
                        text = textMessage.getText();
                        // 第八步：列印訊息。
                        System.out.println(text);
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            }
        });
        // 系統等待接收訊息：
        // 方式一：
        /*      
        while (true) {
            Thread.sleep(100);
        }*/
        // 方式二：等待鍵盤輸入
        System.in.read();
        // 第九步：關閉資源
        consumer.close();
        session.close();
        connection.close();
    }

4.2、Topic（話題）

ActiveMQ的Topic方式與ActiveMQ的Queue方式大部分都是一樣的。
ActiveMQ的Topic方式預設在ActiveMQ的服務端是不持久化的。如果消費者客戶端沒有接受到訊息內容，訊息內容就會丟了。所以消費者客戶端需要先在啟動狀態（Topic方式可以有多個消費者）。我們也可以修改配置將Topic方式持久化快取。後面可以自己學習。

4.2.1、Producer

使用步驟：
第一步：建立ConnectionFactory物件，需要指定服務端ip及埠號。
第二步：使用ConnectionFactory物件建立一個Connection物件。
第三步：開啟連線，呼叫Connection物件的start方法。
第四步：使用Connection物件建立一個Session物件。
第五步：使用Session物件建立一個Destination物件（topic、queue），此處建立一個Topic物件。
第六步：使用Session物件建立一個Producer物件。
第七步：建立一個Message物件，建立一個TextMessage物件。
第八步：使用Producer物件傳送訊息。
第九步：關閉資源。
測試程式碼如下：

    /**
     * ActiveMQ的Topic方式的Consumer
     * @throws Exception
     */
    @Test
    public void topicProducerTest() throws Exception {
        // 第一步：建立ConnectionFactory物件，需要指定服務端ip及埠號61616。注意：8161是ActiveMQ的web服務的埠。
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.168:61616"); // 注意：虛擬機器上的安裝ActiveMQ服務要新增該埠
        // 第二步：使用ConnectionFactory物件建立一個Connection物件。
        Connection connection = connectionFactory.createConnection();
        // 第三步：開啟連線，呼叫Connection物件的start方法。
        connection.start();
        // 第四步：使用Connection物件建立一個Session物件。
        // 第一個引數：是否開啟分散式事務。true：開啟分散式事務，當第一個引數為true時，第二個引數忽略。
        // 一般我們不使用分散式事務，因為分散式事務複雜，效能不好。網際網路專案要求快速響應給使用者，其餘的事情交給訊息佇列向各個服務發訊息讓他們去做。這樣保證了資料的最終一致。
        // 第二個引數：當第一個引數為false時，第二個引數才有意義，表示訊息的應答模式。1、自動應答  2、手動應答。一般是自動應答。
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 第五步：使用Session物件建立一個Destination物件（topic、queue），此處建立一個Topic物件。引數表示話題的名稱。
        Topic topic = session.createTopic("test-topic");
        // 第六步：使用Session物件建立一個Producer物件。
        MessageProducer producer = session.createProducer(topic);
        // 第七步：建立一個Message物件，建立一個TextMessage物件。
        // 方式一：
        // TextMessage message = new ActiveMQTextMessage();
        // message.setText("hello ActiveMQ, this is my first test.");
        // 方式二：
        TextMessage textMessage = session.createTextMessage("hello ActiveMQ, this is my first test.");
        // 第八步：使用Producer物件傳送訊息。
        producer.send(textMessage);
        // 第九步：關閉資源。
        producer.close();
        session.close();
        connection.close();
    }

4.2.2、Consumer

消費者：接收訊息。
第一步：建立一個ConnectionFactory物件。
第二步：從ConnectionFactory物件中獲得一個Connection物件。
第三步：開啟連線。呼叫Connection物件的start方法。
第四步：使用Connection物件建立一個Session物件。
第五步：使用Session物件建立一個Destination物件。和傳送端保持一致topic，並且話題的名稱一致。
第六步：使用Session物件建立一個Consumer物件。
第七步：接收訊息。
第八步：列印訊息。
第九步：關閉資源。
測試程式碼如下：

    /**
     * ActiveMQ的Topic方式的Consumer
     * @throws Exception
     */
    @Test
    public void topicConsumerTest() throws Exception {
        // 第一步：建立一個ConnectionFactory物件。
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.168:61616");
        // 第二步：從ConnectionFactory物件中獲得一個Connection物件。
        Connection connection = connectionFactory.createConnection();
        // 第三步：開啟連線。呼叫Connection物件的start方法。
        connection.start();
        // 第四步：使用Connection物件建立一個Session物件。
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 第五步：使用Session物件建立一個Destination物件。和傳送端保持一致Topic，並且佇列的名稱一致。
        Topic topic = session.createTopic("test-topic");
        // 第六步：使用Session物件建立一個Consumer物件。
        MessageConsumer consumer = session.createConsumer(topic);
        // 第七步：接收訊息。
        consumer.setMessageListener(new MessageListener() {

            @Override
            public void onMessage(Message message) {
                if (message instanceof TextMessage) {
                    TextMessage textMessage = (TextMessage) message;
                    try {
                        String text = null;
                        // 取訊息的內容
                        text = textMessage.getText();
                        // 第八步：列印訊息。
                        System.out.println(text);
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            }
        });
        System.out.println("topic的消費者客戶端01...");
        // 系統等待接收訊息：
        // 方式一：
        /*      
        while (true) {
            Thread.sleep(100);
        }*/
        // 方式二：等待鍵盤輸入
        System.in.read();
        // 第九步：關閉資源
        consumer.close();
        session.close();
        connection.close();
    }

注意：
  Topic 預設是不存在於MQ伺服器中的，一旦傳送之後，如果沒有訂閱，就沒了。
  Queue 預設是存在於MQ的伺服器中的，傳送訊息之後，隨時取。但是一定是一個消費者取完就沒了。

回到頂部

5、ActiveMQ整合spring

5.1、使用方法

第一步：把Activemq模板相關的jar包新增到工程中。在Maven工程中是新增依賴jar包。最開始配置spring時已經配置過了。

第二步：編寫配置檔案applicationContext-activemq.xml。
第三步：配置生產者。使用JMSTemplate物件。傳送訊息。
第四步：在spring容器中配置Destination。

<?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">

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

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

    <!-- 接收和傳送訊息時使用的類 -->
    <!-- 配置訊息的生產者 -->
    <!-- Spring提供的JMS工具類，它可以進行訊息傳送、接收等 -->
    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
        <!-- 這個connectionFactory對應的是我們定義的Spring提供的那個ConnectionFactory物件 -->
        <property name="connectionFactory" ref="connectionFactory"></property>
    </bean>

    <!-- 配置訊息的Destination物件 -->
    <!--這個是佇列目的地，點對點的 -->
    <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
        <constructor-arg name="name" value="spring-test-queue"></constructor-arg>
    </bean> 
    <!-- 這個是話題目的地，一對多的 -->
    <bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic">
        <constructor-arg name="name" value="spring-test-topic"></constructor-arg>
    </bean> 
</beans>

第五步：程式碼測試。

5.2、程式碼測試（Queue方式）

5.2.1、傳送訊息

taotao-manager-service中傳送訊息。
第一步：初始化一個spring容器。即載入配置檔案applicationContext-activemq.xml。
第二步：從容器中獲得JMSTemplate物件。
第三步：從容器中獲得一個Destination物件。
第四步：使用JMSTemplate物件傳送訊息，需要知道Destination。
測試程式碼如下：

    /**
     * ActiveMQ整合spring，ActiveMQ的Queue方式的Producer
     * @throws Exception
     */
    @Test
    public void queueProducerTest() throws Exception {
        // 第一步：初始化一個spring容器，即載入配置檔案applicationContext-activemq.xml。
        ApplicationContext applicationContext = new ClassPathXmlApplicationContext("classpath:spring/applicationContext-activemq.xml");
        // 第二步：從容器中獲得JMSTemplate物件。
        JmsTemplate jmsTemplate = applicationContext.getBean(JmsTemplate.class);
        // 第三步：從容器中獲得一個Destination物件
        Queue queue = (Queue) applicationContext.getBean("queueDestination"); // 建議根據id取
        // 第四步：使用JMSTemplate物件傳送訊息，需要知道Destination
        jmsTemplate.send(queue, new MessageCreator() {

            @Override
            public Message createMessage(Session session) throws JMSException {
                TextMessage textMessage = session.createTextMessage("spring activemq test");
                return textMessage;
            }
        });
    }

5.2.2、接收訊息

taotao-search-service中接收訊息。
第一步：把jar包新增到工程中。使用5.11.2版本的jar包。版本號我們在taotao-parent中的pom.xml中配置。

第二步：建立一個MessageListener的實現類。用於接收ActiveMQ傳送的訊息。

/**
 * 自定義的MessageListener的實現類，用於接收ActiveMQ傳送的訊息
 * @author chenmingjun
 * @date 2018年11月25日下午11:30:56
 * @version 1.0
 */
public class MyMessageListener implements MessageListener {

    @Override
    public void onMessage(Message message) {
        // 接收到訊息，進行消費
        try {
            TextMessage textMessage = (TextMessage) message;
            // 取訊息內容
            String text = textMessage.getText();
            System.out.println(text);
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}

第三步：編寫配置檔案applicationContext-activemq.xml，配置spring和Activemq整合，配置訊息的消費者。

<?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">

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

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

    <!-- 接收和傳送訊息時使用的類 -->
    <!-- 配置訊息的Destination物件 -->
    <!--這個是佇列目的地，點對點的 -->
    <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
        <constructor-arg name="name" value="spring-test-queue"></constructor-arg>
    </bean> 
    <!-- 這個是話題目的地，一對多的 -->
    <bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic">
        <constructor-arg name="name" value="spring-test-topic"></constructor-arg>
    </bean> 

    <!-- 配置訊息的消費者 -->
    <!-- 先配置監聽器 -->
    <bean id="myMessageListener" class="com.taotao.search.listener.MyMessageListener" />
    <!-- 再配置訊息監聽容器 -->
    <bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="connectionFactory" />
        <property name="destination" ref="queueDestination" />
        <property name="messageListener" ref="myMessa