消息中間件ActiveMQ及Spring整合JMS

一 、消息中間件的基本介紹

1.1 消息中間件

1.1.1 什么是消息中間件

消息中間件利用高效可靠的消息傳遞機制進行平台無關的數據交流，並基於數據通信來進行分布式系統的集成。通過提供消息傳遞和消息排隊模型，它可以在分布式環境下擴展進程間的通信。對於消息中間件，常見的角色大致也就有Producer（生產者）、Consumer（消費者）

常見的消息中間件產品：

（1）ActiveMQ

ActiveMQ 是Apache出品，最流行的，能力強勁的開源消息總線。ActiveMQ 是一個完全支持JMS1.1和J2EE 1.4規范的 JMS Provider實現。這里介紹的是ActiveMQ的使用。

（2）RabbitMQ

AMQP協議的領導實現，支持多種場景。淘寶的MySQL集群內部有使用它進行通訊，OpenStack開源雲平台的通信組件，最先在金融行業得到運用。

（3）ZeroMQ

史上最快的消息隊列系統

（4）Kafka

Apache下的一個子項目 。特點：高吞吐，在一台普通的服務器上既可以達到10W/s的吞吐速率；完全的分布式系統。適合處理海量數據。

1.1.2 使用ActiveMQ的場景

1. 進程間通訊，比如在分布式系統中。

2. 解耦，比如像我們公司有許多開發團隊，每個團隊負責業務的不同模塊，各個開發團隊可以使用MQ來通信。

3. 在一些高並發場景下，使用MQ的異步特性。

4. (歡迎補充)

 1.2 JMS簡介

1.2.1什么是JMS

　　JMS（Java Messaging Service）是Java平台上有關面向消息中間件的技術規范，它便於消息系統中的Java應用程序進行消息交換,並且通過提供標准的產生、發送、接收消息的接口簡化企業應用的開發。

       JMS本身只定義了一系列的接口規范，是一種與廠商無關的 API，用來訪問消息收發系統。它類似於 JDBC(Java Database Connectivity)：這里，JDBC 是可以用來訪問許多不同關系數據庫的 API，而 JMS 則提供同樣與廠商無關的訪問方法，以訪問消息收發服務。許多廠商目前都支持 JMS，包括 IBM 的 MQSeries、BEA的 Weblogic JMS service和 Progress 的 SonicMQ，這只是幾個例子。 JMS 使您能夠通過消息收發服務（有時稱為消息中介程序或路由器）從一個 JMS 客戶機向另一個 JML 客戶機發送消息。消息是 JMS 中的一種類型對象，由兩部分組成：報頭和消息主體。報頭由路由信息以及有關該消息的元數據組成。消息主體則攜帶着應用程序的數據或有效負載。

JMS 定義了五種不同的消息正文格式，以及調用的消息類型，允許你發送並接收以一

些不同形式的數據，提供現有消息格式的一些級別的兼容性。

· TextMessage--一個字符串對象

· MapMessage--一套名稱-值對

· ObjectMessage--一個序列化的 Java 對象

· BytesMessage--一個字節的數據流

· StreamMessage -- Java 原始值的數據流

1.2.2 JMS消息傳遞類型

對於消息的傳遞有兩種類型：

一種是點對點的，即一個生產者和一個消費者一一對應；

另一種是發布//訂閱模式，即一個生產者產生消息並進行發送后，可以由多個消費者進行接收。

1.3ActiveMQ下載與安裝

1.3.1下載

官方網站下載：http://activemq.apache.org/

1.3.2安裝（Linux）

（1）將apache-activemq-5.12.0-bin.tar.gz 上傳至服務器【將安裝包移動到所有軟件安裝目錄(個人習慣)  /export/software】

（2）解壓此文件

tar  -zxvf  apache-activemq-5.12.0-bin.tar.gz -C /export/servers

（3）為apache-activemq-5.12.0目錄賦權

chmod 777 apache-activemq-5.12.0

（4） 進入apache-activemq-5.12.0\bin目錄

（5）賦與執行權限

chmod 755 activemq 

Linux下賦值權限的介紹：

linux 命令chmod 755的意思
chmod是Linux下設置文件權限的命令，后面的數字表示不同用戶或用戶組的權限。
一般是三個數字：
第一個數字表示文件所有者的權限
第二個數字表示與文件所有者同屬一個用戶組的其他用戶的權限
第三個數字表示其它用戶組的權限。
      權限分為三種：讀（r=4），寫（w=2），執行（x=1） 。 綜合起來還有可讀可執行（rx=5=4+1）、可讀可寫（rw=6=4+2）、可讀可寫可執行(rwx=7=4+2+1)。
      所以，chmod 755 設置用戶的權限為： 
1.文件所有者可讀可寫可執行                                    --7
2.與文件所有者同屬一個用戶組的其他用戶可讀可執行 --5 
3.其它用戶組可讀可執行                                          --5

1.3.3啟動

 ./activemq start

出現下列提示表示成功！

 

我的服務器地址是192.168.75.14，打開瀏覽器輸入地址http://192.168.75.14:8161/ 即可進入ActiveMQ管理頁面

點擊進入管理頁面

輸入用戶名和密碼  均為 admin 

進入主界面

點擊進入消息列表：

列表各列信息含義如下：

Number Of Pending Messages  ：等待消費的消息 這個是當前未出隊列的數量。

Number Of Consumers  ：消費者 這個是消費者端的消費者數量

Messages Enqueued  ：進入隊列的消息  進入隊列的總數量,包括出隊列的。

Messages Dequeued  ：出了隊列的消息  可以理解為是消費這消費掉的數量。

2. JMS入門小Demo

2.1點對點模式

點對點的模式主要建立在一個隊列上面，當連接一個列隊的時候，發送端不需要知道接收端是否正在接收，可以直接向ActiveMQ發送消息，發送的消息，將會先進入隊列中，如果有接收端在監聽，則會發向接收端，如果沒有接收端接收，則會保存在activemq服務器，直到接收端接收消息，點對點的消息模式可以有多個發送端，多個接收端，但是一條消息，只會被一個接收端給接收到，哪個接收端先連上ActiveMQ，則會先接收到，而后來的接收端則接收不到那條消息。

2.1.1消息生產者

（1）創建工程jmsDemo ，引入依賴

     <dependency>
        <groupId>org.apache.activemq</groupId>
        <artifactId>activemq-client</artifactId>
        <version>5.13.4</version>
     </dependency>

（2）創建類QueueProducer  main方法代碼如下：

//1.創建連接工廠
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.75.14:61616");
        //2.獲取連接
        Connection connection = connectionFactory.createConnection();
        //3.啟動連接
        connection.start();
        //4.獲取session(參數1：是否啟動事務,參數2：消息確認模式)
        Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
        //5.創建隊列對象
        Queue queue = session.createQueue("test-queue");
        //6.創建消息生產者
        MessageProducer producer = session.createProducer(queue);
        //7.創建消息
        TextMessage textMessage = session.createTextMessage("歡迎來到activeMQ的世界!");
        //8.發送消息
        producer.send(textMessage);
        //9.關閉資源
        producer.close();
        session.close();
        connection.close();

上述代碼中第4步創建session  的兩個參數：

第1個參數 是否使用事務

第2個參數 消息的確認模式

• AUTO_ACKNOWLEDGE = 1    自動確認
• CLIENT_ACKNOWLEDGE = 2    客戶端手動確認   
• DUPS_OK_ACKNOWLEDGE = 3    自動批量確認
• SESSION_TRANSACTED = 0    事務提交並確認

 運行后通過ActiveMQ管理界面查詢

2.1.2消息消費者

創建類QueueConsumer ，main方法代碼如下：

//1.創建連接工廠
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.75.14:61616");
        //2.獲取連接
        Connection connection = connectionFactory.createConnection();
        //3.啟動連接
        connection.start();
        //4.獲取session(參數1：是否啟動事務,參數2：消息確認模式)
        Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
        //5.創建隊列對象
        Queue queue = session.createQueue("test-queue");
        //6.創建消息消費
        MessageConsumer consumer = session.createConsumer(queue);
        //7.監聽消息
        consumer.setMessageListener(new MessageListener() {
            
            public void onMessage(Message message) {
                TextMessage textMessage = (TextMessage) message;
                try {
                    System.out.println("接收到消息："+textMessage.getText());
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        });
        //8.等待鍵盤輸入
        System.in.read();    
        //9.關閉資源
        consumer.close();
        session.close();
        connection.close();
    }

執行后看到控制台輸出

2.1.3運行測試

同時開啟2個以上的消費者，再次運行生產者，觀察每個消費者控制台的輸出，會發現只有一個消費者會接收到消息。

 

2.2 發布/訂閱模式

2.2.1消息生產者

創建類TopicProducer ，main方法代碼如下：

//1.創建連接工廠
        ConnectionFactory connectionFactory=new ActiveMQConnectionFactory("tcp://192.168.75.14:61616");
        //2.獲取連接
        Connection connection = connectionFactory.createConnection();
        //3.啟動連接
        connection.start();
        //4.獲取session  (參數1：是否啟動事務,參數2：消息確認模式)
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //5.創建主題對象
        Topic topic = session.createTopic("test-topic");
        //6.創建消息生產者
        MessageProducer producer = session.createProducer(topic);
        //7.創建消息
        TextMessage textMessage = session.createTextMessage("歡迎來到神奇的activeMQ世界");
        //8.發送消息
        producer.send(textMessage);
        //9.關閉資源
        producer.close();
        session.close();
        connection.close();
    }

運行效果如下：

2.2.2消息消費者

創建類TopicConsumer ,main方法代碼如下：

//1.創建連接工廠
        ConnectionFactory connectionFactory=new ActiveMQConnectionFactory("tcp://192.168.75.14:61616");
        //2.獲取連接
        Connection connection = connectionFactory.createConnection();
        //3.啟動連接
        connection.start();
        //4.獲取session  (參數1：是否啟動事務,參數2：消息確認模式)
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //5.創建主題對象
        //Queue queue = session.createQueue("test-queue");
        Topic topic = session.createTopic("test-topic");
        //6.創建消息消費
        MessageConsumer consumer = session.createConsumer(topic);
        
        //7.監聽消息
        consumer.setMessageListener(new MessageListener() {
            public void onMessage(Message message) {
                TextMessage textMessage=(TextMessage)message;
                try {
                    System.out.println("接收到消息："+textMessage.getText());
                } catch (JMSException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }
        });
        //8.等待鍵盤輸入
        System.in.read();
        //9.關閉資源
        consumer.close();
        session.close();
        connection.close();
    }

2.2.3運行測試

同時開啟2個以上的消費者，再次運行生產者，觀察每個消費者控制台的輸出，會發現每個消費者會接收到消息。

3.Spring整合JMS

3.1 點對點模式

3.1.1消息生產者

（1）創建工程springjms_producer，在POM文件中引入SpringJms 、activeMQ以及單元測試相關依賴 

<dependencies>
          <dependency>
            <groupId>org.apache.activemq</groupId>
            <artifactId>activemq-client</artifactId>
            <version>5.13.4</version>
         </dependency>
          <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-jms</artifactId>    
            <version>4.3.2.RELEASE</version>    
        </dependency>
          <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-test</artifactId>
            <version>4.2.4.RELEASE</version>    
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.9</version>
        </dependency>
  </dependencies>

（2）在src/main/resources下創建spring配置文件(先創建spring文件夾)applicationContext-jms-producer.xml

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" 
    xmlns:p="http://www.springframework.org/schema/p"
    xmlns:context="http://www.springframework.org/schema/context"
    xmlns:dubbo="http://code.alibabatech.com/schema/dubbo" 
    xmlns:mvc="http://www.springframework.org/schema/mvc"
    xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
        http://www.springframework.org/schema/mvc http://www.springframework.org/schema/mvc/spring-mvc.xsd
        http://code.alibabatech.com/schema/dubbo http://code.alibabatech.com/schema/dubbo/dubbo.xsd
        http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
    
   <context:component-scan base-package="com.yida.demo"></context:component-scan>        
    <!-- 真正可以產生Connection的ConnectionFactory，由對應的 JMS服務廠商提供-->  
    <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">  
        <property name="brokerURL" value="tcp://192.168.75.14:61616"/>  
    </bean>          
    <!-- Spring用於管理真正的ConnectionFactory的ConnectionFactory -->  
    <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">  
    <!-- 目標ConnectionFactory對應真實的可以產生JMS Connection的ConnectionFactory -->  
        <property name="targetConnectionFactory" ref="targetConnectionFactory"/>  
    </bean>         
    <!-- Spring提供的JMS工具類，它可以進行消息發送、接收等 -->  
    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">  
        <!-- 這個connectionFactory對應的是我們定義的Spring提供的那個ConnectionFactory對象 -->  
        <property name="connectionFactory" ref="connectionFactory"/>  
    </bean>      
    <!--這個是隊列目的地，點對點的  文本信息-->  
    <bean id="queueTextDestination" class="org.apache.activemq.command.ActiveMQQueue">  
        <constructor-arg value="queue_text"/>  
    </bean>  
       

</beans>

（3）在com.yida.demo包下創建消息生產者類

@Component
public class QueueProducer {
    
    @Autowired
    private JmsTemplate jmsTemplate;
    
    @Autowired
    private Destination queueTextDestination;//配置文件applicationContext-jms-producer.xml點對點文本信息bean的id
    
    /**
     * 發送文本消息
     * @param text
     */
    public void sendTextMessage(final String text){
        jmsTemplate.send(queueTextDestination, new MessageCreator() {
            
            public Message createMessage(Session session) throws JMSException {
                return session.createTextMessage(text);
            }
        });
    }
    
}

（4）單元測試

在src/test/java創建測試類

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations="classpath:spring/applicationContext-jms-producer.xml")
public class TestQueue {
    @Autowired
    private QueueProducer queueProducer;
    
    @Test
    public void testSend(){
        queueProducer.sendTextMessage("SpringJms-點對點");
    }
}

3.1.2消息消費者

（1）創建工程springjms_consumer，在POM文件中引入依賴 （同上一個工程）

（2）創建配置文件 applicationContext-jms-consumer-queue.xml

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" 
    xmlns:p="http://www.springframework.org/schema/p"
    xmlns:context="http://www.springframework.org/schema/context"
    xmlns:dubbo="http://code.alibabatech.com/schema/dubbo" 
    xmlns:mvc="http://www.springframework.org/schema/mvc"
    xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
        http://www.springframework.org/schema/mvc http://www.springframework.org/schema/mvc/spring-mvc.xsd
        http://code.alibabatech.com/schema/dubbo http://code.alibabatech.com/schema/dubbo/dubbo.xsd
        http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
    
   <!-- 真正可以產生Connection的ConnectionFactory，由對應的 JMS服務廠商提供-->  
    <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">  
        <property name="brokerURL" value="tcp://192.168.75.14:61616"/>  
    </bean>       
    <!-- Spring用於管理真正的ConnectionFactory的ConnectionFactory -->  
    <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">  
    <!-- 目標ConnectionFactory對應真實的可以產生JMS Connection的ConnectionFactory -->  
        <property name="targetConnectionFactory" ref="targetConnectionFactory"/>  
    </bean>      
    <!--這個是隊列目的地，點對點的  文本信息-->  
    <bean id="queueTextDestination" class="org.apache.activemq.command.ActiveMQQueue">  
        <constructor-arg value="queue_text"/>  
    </bean>    
    <!-- 我的監聽類 -->
    <bean id="myMessageListener" class="com.yida.demo.MyMessageListener"></bean>
    <!-- 消息監聽容器 -->
    <bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="connectionFactory" />
        <property name="destination" ref="queueTextDestination" />
        <property name="messageListener" ref="myMessageListener" />
    </bean>    
   
       

</beans>

（3）編寫監聽類

public class MyMessageListener implements MessageListener {

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

（4）創建測試類

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations="classpath:spring/applicationContext-jms-consumer-queue.xml")
public class TestQueue {
    @Test
    public void testQueue(){
        try {
            System.in.read();
        } catch (IOException e) {
            e.printStackTrace();
        }        
    }    
}

3.1.3 測試結果

先運行springjms_producer工程的生產者測試類，再運行springjms_consumer工程的測試類 ，控制台打印如下：

3.2發布/訂閱模式

3.2.1消息生產者

（1）在工程springjms_producer的applicationContext-jms-producer.xml增加配置

    <!--這個是訂閱模式  文本信息-->  
    <bean id="topicTextDestination" class="org.apache.activemq.command.ActiveMQTopic">  
        <constructor-arg value="topic_text"/>  
    </bean>  

（2）創建生產者類

@Component
public class TopicProducer {
    @Autowired
    private JmsTemplate jmsTemplate;
    
    @Autowired
    private Destination topicTextDestination;
    
    /**
     * 發送文本消息
     * @param text
     */
    public void sendTextMessage(final String text){
        jmsTemplate.send(topicTextDestination, new MessageCreator() {            
            public Message createMessage(Session session) throws JMSException {
                return session.createTextMessage(text);
            }
        });        
    }
}

（3）編寫測試類

package springjms_producer;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

import com.yida.demo.TopicProducer;

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations="classpath:spring/applicationContext-jms-producer.xml")
public class TestTopic {
    @Autowired
    private TopicProducer topicProducer;
    @Test
    public void sendTextQueue(){        
        topicProducer.sendTextMessage("我是發布訂閱");
    }    
}

3.2.2消息消費者

（1）在springjms_consumer工程中創建配置文件applicationContext-jms-consumer-topic.xml

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" 
    xmlns:p="http://www.springframework.org/schema/p"
    xmlns:context="http://www.springframework.org/schema/context"
    xmlns:dubbo="http://code.alibabatech.com/schema/dubbo" 
    xmlns:mvc="http://www.springframework.org/schema/mvc"
    xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
        http://www.springframework.org/schema/mvc http://www.springframework.org/schema/mvc/spring-mvc.xsd
        http://code.alibabatech.com/schema/dubbo http://code.alibabatech.com/schema/dubbo/dubbo.xsd
        http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
    
     <!-- 真正可以產生Connection的ConnectionFactory，由對應的 JMS服務廠商提供-->  
    <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">  
        <property name="brokerURL" value="tcp://192.168.75.14:61616"/>  
    </bean>
    <!-- Spring用於管理真正的ConnectionFactory的ConnectionFactory -->  
    <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">  
    <!-- 目標ConnectionFactory對應真實的可以產生JMS Connection的ConnectionFactory -->  
        <property name="targetConnectionFactory" ref="targetConnectionFactory"/>  
    </bean>  
    <!--這個是隊列目的地，點對點的  文本信息-->  
    <bean id="topicTextDestination" class="org.apache.activemq.command.ActiveMQTopic">  
        <constructor-arg value="topic_text"/>  
    </bean>    
    <!-- 我的監聽類 -->
    <bean id="myMessageListener" class="com.yida.demo.MyMessageListener"></bean>
    <!-- 消息監聽容器 -->
    <bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="connectionFactory" />
        <property name="destination" ref="topicTextDestination" />
        <property name="messageListener" ref="myMessageListener" />
    </bean>
 
       

</beans>

（2）編寫測試類

package springjms_consumer;

import java.io.IOException;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations="classpath:spring/applicationContext-jms-consumer-topic.xml")
public class TestTopic {
    @Test
    public void testTopic(){
        try {
            System.in.read();
        } catch (IOException e) {
            e.printStackTrace();
        }        
    }        
}

測試：同時運行三個消費者工程，在運行生產者工程，查看三個消費者工程的控制台輸出。