第十一章 企業項目開發--消息隊列activemq

注意：本章代碼基於 第十章 企業項目開發--分布式緩存Redis(2)

代碼的github地址：https://github.com/zhaojigang/ssmm0

 

消息隊列是分布式系統中實現RPC的一種手段。

1、消息隊列的基本使用流程

假設：

• 我們有這樣一個需求，當每注冊一個admin的之后，就寫一條日志log數據到數據庫。

分析：

• 在實際中，我們是不會把日志直接寫入數據庫的，因為日志數據通常是龐大的，而且日志的產生是頻繁的，如果我們使用數據庫存儲日志，哪怕是使用異步存儲，也是極耗性能的。在企業中，對於日志的處理方式很多，比較簡單的一種是，日志直接產生於nginx或后端服務器（eg.resin），我們寫一個定時任務，每隔一段時間，將產生的日志文件使用shell和Python進行正則過濾，取出有用信息，之后進行處理統計，最后將處理后的數據寫入數據庫。

在這里我們作為演示，，每當注冊一個admin的之后，我們異步寫一條日志log數據到數據庫。

下邊的舉例也是對代碼的解釋。

• server1：部署ssmm0-userManagement
• server2：部署ssmm0-rpcWeb
• server3：部署消息隊列服務器

當server1執行一個"http://localhost:8080/admin/register?username=canglang25&password=1457890"操作，即向數據庫插一條admin信息時，同時將日志log信息寫入server3，之后不會等待log信息被server2消費掉就直接返回（異步）；

server2循環接收server3中的消息隊列中的消息，並將這些log消息寫入數據庫。

 

2、消息隊列的作用

• 異步
• 解耦：server1（消息生產者服務器）和server3（消息消費者服務器）沒有直接聯系
• 削峰填谷：當大量請求涌入應用服務器時，應用服務器如果處理不過來，就將這些請求先放入隊列，之后再從隊列中取出請求慢慢處理（秒殺的一種處理方式）

 

3、消息隊列的兩種方式

• P2P
  • 消息生產者產生的消息只能由一個消息消費者消費
  • 基於隊列queue
  • 執行流程
    • 生產者：創建連接工廠-->創建連接-->啟動連接-->創建session-->創建隊列，創建生產者，創建消息-->發送消息
    • 消費者：創建連接工廠-->創建連接-->啟動連接-->創建session-->創建隊列，創建消費者-->接收消息
• 發布-訂閱
  • 消息生產者產生的消息可以由所有訂閱了（監聽了）該消息的消費者消費
  • 基於主題topic
  • 執行流程
    • 生產者：創建連接工廠-->創建連接-->啟動連接-->創建session-->創建topic，創建消息發布者，創建消息-->發布消息
    • 消費者：創建連接工廠-->創建連接-->啟動連接-->創建session-->創建topic，創建消息訂閱者-->消息訂閱者通過監聽器接收消息

 

4、實例（基於P2P實現）

4.1、整體代碼結構：

 

4.2、模塊依賴關系

注：箭頭的指向就是當前模塊所依賴的模塊。（eg.rpcWeb依賴data）

• userManagement：用戶管理模塊--war
• rpcWeb：rpc測試模塊（這里用於模擬接收處理消息的應用）--war
• cache：緩存模塊--jar
• rpc：rpc模塊（包含mq/mina/netty）--jar
• data：數據處理模塊--jar
• common：通用工具類模塊--jar

 

4.3、代碼

代碼整體沒變，只列出部分新增代碼，完整代碼從文首的github進行clone即可。

4.3.1、ssmm0

pom.xml

    <!-- 管理子模塊 -->
    <modules>
        <module>common</module><!-- 通用類模塊 -->
        <module>cache</module><!-- 緩存模塊 -->
        <module>rpc</module><!-- rpc模塊 -->
        <module>data</module><!-- 封裝數據操作 -->
        <module>userManagement</module><!-- 具體業務1-人員管理系統，這里的userManagement部署在serverA上（配合rpcWeb測試rpc） -->
        <module>rpcWeb</module><!-- 具體業務2-用於測試RPC的另一台機器，這里的rpcWeb項目部署在serverB上 -->
    </modules>

            <!-- 日志：若沒有，activemq獲取連接報錯 -->
            <dependency>
                <groupId>org.slf4j</groupId>
                <artifactId>slf4j-log4j12</artifactId>
                <version>1.5.11</version>
            </dependency>

說明：只列出部分新增的代碼。

注意：

• activemq必須配置slf4j-log4j12，而該jar也會被所有的模塊用到（因為所有的模塊都需要打日志），至於該模塊的版本號的選擇我們可以根據"啟動activemq，並運行自己的程序"從eclipse的console窗口的打印信息來選擇。
• slf4j-log4j12這個jar在pom.xml中引入到依賴池中后，還需要進行實際依賴
• module部分最好按照依賴關系從底向上排列，這樣在"compile"的時候不容易出錯

 

4.3.2、ssmm0-common

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">

    <modelVersion>4.0.0</modelVersion>

    <!-- 指定父模塊 -->
    <parent>
        <groupId>com.xxx</groupId>
        <artifactId>ssmm0</artifactId>
        <version>1.0-SNAPSHOT</version>
    </parent>

    <groupId>com.xxx.ssmm0</groupId>
    <artifactId>ssmm0-common</artifactId>

    <name>ssmm0-common</name>
    <packaging>jar</packaging>
    
    <dependencies>
    <!-- bc-加密 -->
        <dependency>
            <groupId>org.bouncycastle</groupId>
            <artifactId>bcprov-jdk15on</artifactId>
        </dependency>
        <!-- cc加密 -->
        <dependency>
            <groupId>commons-codec</groupId>
            <artifactId>commons-codec</artifactId>
        </dependency>
    </dependencies>
</project>

DateUtil：

package com.xxx.util;

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

/**
 * 線程安全的日期類工具
 */
public class DateUtil {
    private static final String DATE_FORMAT = "yyyy-MM-dd HH:mm:ss";
    private static ThreadLocal<DateFormat> threadLocal = new ThreadLocal<DateFormat>();
    
    /**
     * 獲取DateFormat實例
     */
    public static DateFormat getDateFormat() {
        DateFormat df = threadLocal.get();//從threadLocal中獲取當前線程的DateFormat實例副本
        if(df==null){//如果當前線程實例為null，說明該線程第一次使用該方法
            df = new SimpleDateFormat(DATE_FORMAT);//創建df實例
            threadLocal.set(df);//將df實例放置到threadLocal中去
        }
        return df;
    }
    
    /**
     * 將Date格式化為String字符串
     */
    public static String formatDate(Date date) {
        return getDateFormat().format(date);
    }
    
    /**
     * 獲取當前時間
     * @return 字符串（eg.2001-11-12 12:23:34）
     */
    public static String getCurrentTime() {
        //第一種方式
        //return formatDate(new Date());
        
        //第二種方式（也是最推薦的方式）
        DateFormat df = getDateFormat();
        return df.format(System.currentTimeMillis());
        
        //第三種方式
        /*Calendar c = Calendar.getInstance();
        return c.get(Calendar.YEAR)+"-"+c.get(Calendar.MONTH)+"-"+c.get(Calendar.DATE)
            +"-"+c.get(Calendar.HOUR)+"-"+c.get(Calendar.MINUTE)+"-"+c.get(Calendar.SECOND);*/
    }
    
    /*****************測試*****************/
    /*public static void main(String[] args) {
        System.out.println(getCurrentTime());
    }*/
}

注意：

• jdk的SimpleDateFormat類是一個線程不安全的類，一般情況下只要不設置為static型類變量就可以了，但是更安全的做法是使用ThreadLocal類包裝一下（如代碼所示），當然也可以使用其他的日期工具。
• 獲取當前時間有三種方式（如代碼所示），最推薦的是第二種

PropUtil：即之前的FileUtil

 

4.3.3、ssmm0-rpc

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">

    <modelVersion>4.0.0</modelVersion>

    <!-- 指定父模塊 -->
    <parent>
        <groupId>com.xxx</groupId>
        <artifactId>ssmm0</artifactId>
        <version>1.0-SNAPSHOT</version>
    </parent>

    <groupId>com.xxx.ssmm0</groupId>
    <artifactId>ssmm0-rpc</artifactId>

    <name>ssmm0-rpc</name>
    <packaging>jar</packaging>

    <!-- 引入實際依賴 -->
    <dependencies>
        <!-- 引入自定義common模塊 -->
        <dependency>
            <groupId>com.xxx.ssmm0</groupId>
            <artifactId>ssmm0-common</artifactId>
            <version>1.0-SNAPSHOT</version>
        </dependency>
        <!-- activemq -->
        <dependency>
            <groupId>org.apache.activemq</groupId>
            <artifactId>activemq-all</artifactId>
            <version>5.5.0</version>
        </dependency>
    </dependencies>
</project>

rpc_config.properties

#activemq配置
activemq.queueURL=tcp://127.0.0.1:61616
activemq.queueName=adminQueue

說明：

• 這里直接將數據配置在這里了，實際上可以將數據配置到ssmm0的根pom.xml中去。

ActiveMQP2PUtil：基於P2P的activemq的消息收發工具類

package com.xxx.rpc.mq.util;

import java.io.Serializable;
import java.util.Properties;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

import com.xxx.rpc.mq.handler.MessageHandler;
import com.xxx.util.PropUtil;

/**
 * activemq p2p 工具類 
 */
public class ActiveMQP2PUtil {
    private static final String RPC_CONFIG_FILE = "rpc_config.properties";
    private static String queueURL;        //隊列所在的URL
    private static String queueName;    //隊列名稱
    private static ConnectionFactory connectionFactory;    //連接工廠
    
    static{
        Properties props = PropUtil.loadProps(RPC_CONFIG_FILE);
        queueURL = props.getProperty("activemq.queueURL", "tcp://127.0.0.1:61616");
        System.out.println(queueURL);
        queueName = props.getProperty("activemq.queueName", "adminQueue");
        connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER, 
                                                          ActiveMQConnection.DEFAULT_PASSWORD, 
                                                          queueURL);
    }
    
    /**
     * 發送消息
     */
    public static void sendMessage(Serializable message){
        Connection conn = null;
        try {
            conn = connectionFactory.createConnection();//創建連接
            conn.start();//啟動連接
            Session session = conn.createSession(true, Session.AUTO_ACKNOWLEDGE);//創建session
            Destination destination = session.createQueue(queueName);//創建隊列
            MessageProducer producer = session.createProducer(destination);
            producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);//消息設置為非持久化
            ObjectMessage msg = session.createObjectMessage(message);//創建消息：createObjectMessage()該方法的入參是Serializable型的
            producer.send(msg);//發送消息
            session.commit();//提交消息
        } catch (JMSException e) {
            e.printStackTrace();
        }finally{
            if(conn!=null){
                try {
                    conn.close();
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        }
    }
    
    /**
     * 接收消息
     * @param handler 自定義的消息處理器
     */
    public static void receiveMessage(MessageHandler handler){
        Connection conn = null;
        try {
            conn = connectionFactory.createConnection();//創建連接
            conn.start();//啟動連接
            Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);//創建session
            Destination destination = session.createQueue(queueName);//創建隊列
            MessageConsumer consumer = session.createConsumer(destination);//創建消息消費者
            while(true){//死循環接收消息
                Message msg = consumer.receive();//接收消息
                if(msg!=null){
                    handler.handle(msg);//處理消息
                    //System.out.println(msg);
                }
            }
        } catch (JMSException e) {
            e.printStackTrace();
        }finally{
            if(conn!=null){
                try {
                    conn.close();
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        }
    }
    
    /*public static void main(String[] args) {
        sendMessage("hello world3");
    }*/
}

說明：

• 對照P2P的執行流程來看代碼
• 關於static塊的執行時機，可以去看 第四章 類加載機制
  • 在我們啟動spring容器時，上述的static塊不執行，只有第一次使用到該類的時候才執行
  • 假設我們為該類添加了注解@Component，那么該類會由spring容器來管理，在spring初始化bean之后就會執行該static塊（也就是說spring容器啟動時，執行static塊）
  • 若將該類不添加如上注解，直接實現接口InitializingBean，並且將static代碼塊中的信息寫到afterPropertiesSet()方法中，則spring容器啟動時，執行static塊
• 對於消息的接收，這里采用了循環等待機制（即死循環），也可以使用事件通知機制
• 關於activemq的其他內容之后再說

MessageHandler：消息處理器接口（其實現類是對接收到的消息進行處理的真正部分）

package com.xxx.rpc.mq.handler;

import javax.jms.Message;

/**
 * 消息處理器接口
 */
public interface MessageHandler {
    public void handle(Message message);
}

 

4.3.4、ssmm0-data

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">

    <modelVersion>4.0.0</modelVersion>

    <!-- 指定父模塊 -->
    <parent>
        <groupId>com.xxx</groupId>
        <artifactId>ssmm0</artifactId>
        <version>1.0-SNAPSHOT</version>
    </parent>

    <groupId>com.xxx.ssmm0</groupId>
    <artifactId>ssmm0-data</artifactId>

    <name>ssmm0-data</name>
    <packaging>jar</packaging><!-- 只是作為其他模塊使用的工具 -->

    <!-- 引入實際依賴 -->
    <dependencies>
        <!-- mysql -->
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
        </dependency>
        <!-- 數據源 -->
        <dependency>
            <groupId>org.apache.tomcat</groupId>
            <artifactId>tomcat-jdbc</artifactId>
        </dependency>
        <!-- mybatis -->
        <dependency>
            <groupId>org.mybatis</groupId>
            <artifactId>mybatis</artifactId>
        </dependency>
        <dependency>
            <groupId>org.mybatis</groupId>
            <artifactId>mybatis-spring</artifactId>
        </dependency>
        <!-- servlet --><!-- 為了會用cookie -->
        <dependency>
            <groupId>javax.servlet</groupId>
            <artifactId>javax.servlet-api</artifactId>
        </dependency>
        <!-- guava cache -->
        <dependency>
            <groupId>com.google.guava</groupId>
            <artifactId>guava</artifactId>
            <version>14.0.1</version>
        </dependency>
        <!-- 引入自定義cache模塊 -->
        <dependency>
            <groupId>com.xxx.ssmm0</groupId>
            <artifactId>ssmm0-cache</artifactId>
            <version>1.0-SNAPSHOT</version>
        </dependency>
        <!-- 引入自定義rpc模塊 -->
        <dependency>
            <groupId>com.xxx.ssmm0</groupId>
            <artifactId>ssmm0-rpc</artifactId>
            <version>1.0-SNAPSHOT</version>
        </dependency>
    </dependencies>
</project>

Log：日志模型類

package com.xxx.model.log;

import java.io.Serializable;

/**
 * 日志
 */
public class Log implements Serializable {
    private static final long serialVersionUID = -8280602625152351898L;
    
    private String operation;   // 執行的操作
    private String currentTime; // 當前時間

    public String getOperation() {
        return operation;
    }

    public void setOperation(String operation) {
        this.operation = operation;
    }

    public String getCurrentTime() {
        return currentTime;
    }

    public void setCurrentTime(String currentTime) {
        this.currentTime = currentTime;
    }
}

注意：

• 需要實現序列化接口，在activemq中的消息需要序列化和反序列化

說明：對應的數據庫表

LogMapper

package com.xxx.mapper.log;

import org.apache.ibatis.annotations.Insert;

import com.xxx.model.log.Log;

/**
 * 日志Mapper
 */
public interface LogMapper {
    
    /**
     * 這里需要注意的是，current_time是數據庫的保留參數，兩點注意：
     * 1、最好不要用保留參數做變量名
     * 2、如果不經意間已經用了，那么保留參數需要用``括起來（`-->該符號是英文狀態下esc鍵下邊的那個鍵）
     * @param log
     * @return
     */
    @Insert("INSERT INTO log(operation, `current_time`) VALUES(#{operation},#{currentTime})")
    public int insertLog(Log log);

}

注意：由於疏忽，在創建數據庫的時候，屬性"當前時間"取名為"current_time"，沒注意到該詞是MySQL的關鍵字（即保留字）。

• 最好不要用關鍵字做變量名
• 如果不經意間已經用了，那么保留參數需要用``括起來（`-->該符號是英文狀態下esc鍵下邊的那個鍵）

LogDao：

package com.xxx.dao.log;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Repository;

import com.xxx.mapper.log.LogMapper;
import com.xxx.model.log.Log;

/**
 * 日志DAO
 */
@Repository
public class LogDao {

    @Autowired
    private LogMapper logMapper;
    /***************注解*****************/
    public boolean insertLog(Log log){
        return logMapper.insertLog(log)==1?true:false;
    }
    
}

LogMessageHandler：MessageHandler的實現類，對接收到的log消息進行具體的操作

package com.xxx.service.log;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.ObjectMessage;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.xxx.dao.log.LogDao;
import com.xxx.model.log.Log;
import com.xxx.rpc.mq.handler.MessageHandler;


/**
 * 日志處理器（更適合放在data層）
 * 因為：
 * 1、data依賴於rpc，而rpc不依賴於data，所以如果該類放在rpc層，並且該類需要用到數據庫操作（eg.將日志寫入數據庫），那么就不好辦了
 * 2、rpc層說白了，就是一些rpc工具類，實際上與業務無關，與業務有關的，我們可以抽取到該部分來
 */
@Component
public class LogMessageHandler implements MessageHandler {

    @Autowired
    private LogDao logDao;
    
    public void handle(Message message) {
        System.out.println(logDao);
        ObjectMessage objMsg = (ObjectMessage)message;
        try {
            Log log = (Log)objMsg.getObject();
            logDao.insertLog(log);//將日志寫入數據庫
        } catch (JMSException e) {
            e.printStackTrace();
        }

    }

}

說明：

• 該類相當於一個service
• 該類放在data模塊而不是rpc模塊，其接口放在了rpc模塊，原因：
  • data依賴於rpc，而rpc不依賴於data，所以如果該類放在rpc層，並且該類需要用到數據庫操作（eg.將日志寫入數據庫），那么就不好辦了
  • rpc層說白了，就是一些rpc工具類，實際上與業務無關，與業務有關的，我們可以抽取到該部分來

AdminService：

    /**
     * 測試activeMQ
     * 
     * 消息生產者做的事：（部署在服務器A）
     * 1）添加一個用戶
     * 2）用戶添加成功后，
     * 2.1）創建一個Log（日志類）實例
     * 2.2）將該日志實例作為消息發送給消息隊列
     * 
     * 消息消費者做的事：（部署在服務器B）
     * 1）從隊列接收消息
     * 2）用日志處理器對消息進行操作（將該消息寫入數據庫）
     */
    public boolean register(Admin admin) {
        boolean isRegisterSuccess = adminDao.register(admin);
        if(isRegisterSuccess) {
            Log log = new Log();
            log.setOperation("增加一個用戶");
            log.setCurrentTime(DateUtil.getCurrentTime());
            
            ActiveMQP2PUtil.sendMessage(log);//將消息發送到消息服務器（即activeMQ服務器），不需要等待消息處理結果，直接向下執行
        }
        return isRegisterSuccess;
    }

說明：

• 該類只修改了以上方法
• 將消息發送到消息服務器（即activeMQ服務器），不需要等待消息處理結果，直接向下執行（體現異步）

 

4.3.5、ssmm0-rpcWeb

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">

    <modelVersion>4.0.0</modelVersion>

    <!-- 指定父模塊 -->
    <parent>
        <groupId>com.xxx</groupId>
        <artifactId>ssmm0</artifactId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    
    <groupId>com.xxx.ssmm0</groupId>
    <artifactId>ssmm0-rpcWeb</artifactId>

    <name>ssmm0-rpcWeb</name>
    <packaging>war</packaging><!-- 需要部署的模塊 -->

    <!-- 引入實際依賴 -->
    <dependencies>
        <!-- 將ssmm0-data項目作為一個jar引入項目中 -->
        <dependency>
            <groupId>com.xxx.ssmm0</groupId>
            <artifactId>ssmm0-data</artifactId>
            <version>1.0-SNAPSHOT</version>
        </dependency>
        <!-- spring mvc(如果沒有web.xml中的CharacterEncodingFilter找不到) -->
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-webmvc</artifactId>
        </dependency>
    </dependencies>
</project>

spring.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:context="http://www.springframework.org/schema/context"
    xmlns:mvc="http://www.springframework.org/schema/mvc"
    xsi:schemaLocation="http://www.springframework.org/schema/beans 
                           http://www.springframework.org/schema/beans/spring-beans-3.2.xsd
                           http://www.springframework.org/schema/context 
                           http://www.springframework.org/schema/context/spring-context-3.2.xsd
                           http://www.springframework.org/schema/mvc http://www.springframework.org/schema/mvc/spring-mvc-3.2.xsd">
                           
    <!-- 注解掃描 -->
    <context:component-scan base-package="com.xxx.web" /><!-- 只掃描web就可以 -->
    
    <!-- 這里需要引入ssmm0-data項目中配置的spring-data.xml(之前不引也可以成功，忘記怎么配置的了) -->
    <import resource="classpath:spring-data.xml"/>
</beans>

web.xml

<?xml version="1.0" encoding="utf-8"?>
<web-app version="2.5" xmlns="http://java.sun.com/xml/ns/javaee"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://java.sun.com/xml/ns/javaee http://java.sun.com/xml/ns/javaee/web-app_2_5.xsd">

    <servlet>
        <servlet-name>dispatcherServlet</servlet-name>
        <servlet-class>org.springframework.web.servlet.DispatcherServlet</servlet-class>
        <init-param>
            <param-name>contextConfigLocation</param-name>
            <param-value>classpath:spring.xml</param-value>
        </init-param>
        <load-on-startup>1</load-on-startup>
    </servlet>
    <servlet-mapping>
        <servlet-name>dispatcherServlet</servlet-name>
        <url-pattern>/</url-pattern>
    </servlet-mapping>

    <filter>
        <filter-name>encodingFilter</filter-name>
        <filter-class>org.springframework.web.filter.CharacterEncodingFilter</filter-class>
        <init-param>
            <param-name>encoding</param-name>
            <param-value>UTF-8</param-value>
        </init-param>
        <init-param>
            <param-name>forceEncoding</param-name>
            <param-value>true</param-value>
        </init-param>
    </filter>
    <filter-mapping>
        <filter-name>encodingFilter</filter-name>
        <url-pattern>/*</url-pattern>
    </filter-mapping>

    <welcome-file-list>
        <welcome-file>/index.jsp</welcome-file>
    </welcome-file-list>
</web-app>

MessageReceiver：死循環從隊列接收消息並將消息傳給消息處理器實現類（LogMessageHandler）處理

package com.xxx.web.mq;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;

import com.xxx.rpc.mq.util.ActiveMQP2PUtil;
import com.xxx.service.log.LogMessageHandler;

/**
 * 用於接收消息的測試類
 */
@Controller
@RequestMapping("/mq")
public class MessageReceiver {
    
    @Autowired
    private LogMessageHandler handler;
    
    @RequestMapping("/receive")
    public void receiveMessage() {
        ActiveMQP2PUtil.receiveMessage(handler);
    }
    
}

• 該類相當於一個controller

 

5、測試

5.1、安裝activemq

1）下載解壓"apache-activemq-5.5.0-bin.zip"，之后，若是32bit機器，進入"E:\activemq-5.5.0\bin\win32"下，雙擊"activemq.bat"即可。（當然，如果雙擊無法啟動，可能有其他進程占用61616端口，查一下是哪一個進程，然后去服務中關掉即可）

2）啟動服務后，在瀏覽器輸入"http://127.0.0.1:8161/admin/queues.jsp"，看到隊列頁面，則安裝並啟動成功，該頁面是一個隊列消息的監控頁面，包括

• 隊列名稱：Name
• 當下有多少消息在隊列中等待消費：Number Of Pending Messages
• 有幾個消費者：Number Of Consumers
• 從啟動activemq服務到現在一共入隊了多少消息：Messages Enqueued
• 從啟動activemq服務到現在一共出隊了多少消息：Messages Dequeued
• Number Of Pending Messages + Messages Dequeued = Messages Enqueued

 

5.2、運行ssmm0-userManagement

瀏覽器執行"http://localhost:8080/admin/register?username=canglang25&password=1457890"

注意：這里使用了8080端口

 

5.3、運行ssmm0-rpcWeb

瀏覽器執行"http://localhost:8081/mq/receive"

注意：

• 這里使用了8081端口
• 執行該URL后，瀏覽器會一直在轉圈（即一直在等待接收消息），直到關閉jetty服務器

 

說明：jetty在不同的端口下可以同時啟動，在同一端口下后邊啟動的服務會覆蓋之前啟動的服務

 

6、總結

• 消息隊列入門簡單，想要完全掌握很難
• 關於git的基本使用查看《progit中文版》