第十一章企业项目开发--消息队列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、整体代码结构：

第十一章企业项目开发--消息队列activemq

4.2、模块依赖关系

第十一章企业项目开发--消息队列activemq

注：箭头的指向就是当前模块所依赖的模块。（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

第十一章企业项目开发--消息队列activemq

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

第十一章企业项目开发--消息队列activemq

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

第十一章企业项目开发--消息队列activemq

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中的消息需要序列化和反序列化

说明：对应的数据库表

第十一章企业项目开发--消息队列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

第十一章企业项目开发--消息队列activemq

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:/admin/register?username=canglang25&password=1457890"

注意：这里使用了8080端口

5.3、运行ssmm0-rpcWeb

浏览器执行"http://localhost:/mq/receive"

注意：

这里使用了8081端口
执行该URL后，浏览器会一直在转圈（即一直在等待接收消息），直到关闭jetty服务器

说明：jetty在不同的端口下可以同时启动，在同一端口下后边启动的服务会覆盖之前启动的服务

6、总结

消息队列入门简单，想要完全掌握很难
关于git的基本使用查看《progit中文版》

秒客网

第十一章企业项目开发--消息队列activemq

相关文章

第十一章 企业项目开发--消息队列activemq

相关文章

第十一章企业项目开发--消息队列activemq