Spring 实现远程访问详解——jms和activemq

た入场券 2022-08-21 06:45 170阅读 0赞

前几章我们分别利用spring rmi、httpinvoker、httpclient、webservice技术实现不同服务器间的远程访问。本章我将通过spring jms和activemq实现单Web项目服务器间异步访问和多Web项目服务器间异步访问。

# 一.  简介 #

## 1.      什么是Apache ActiveMq ##

Apache ActiveMq是最流行和最强大的开源消息和集成服务器。同时Apache ActiveMq是速度快，支持多种跨语言客户端和协议，同时配有易于使用的企业集成模式和优秀的特性，并且支持JMS1.1和J2EE1.4。具体特性见官网：[http://activemq.apache.org/][http_activemq.apache.org]

## 2.      什么是JMS ##

JMS的全称是Java Message Service，即Java消息服务。它主要用于在生产者和消费者之间进行消息传递，生产者负责产生消息，而消费者负责接收消息。把它应用到实际的业务需求中的话我们可以在特定的时候利用生产者生成一消息，并进行发送，对应的消费者在接收到对应的消息后去完成对应的业务逻辑。

JMS 支持两种消息传递模型：

点对点（point-to-point，简称 PTP）

发布/订阅（publish/subscribe，简称 pub/sub）。

这两种消息传递模型非常相似，但有以下区别：

PTP 消息传递模型规定了一条消息只能传递给一个接收方。 采用javax.jms.Queue表示。Spring配置类型destination-type="queue"。

Pub/sub 消息传递模型允许一条消息传递给多个接收方。采用javax.jms.Topic表示。Spring配置类型destination-type="topic"。

# 二.  单服务器异步访问 #

## 3.      Spring 整合JMS和ActiveMq流程 ##

1)      下载和部署ActiveMq服务器

2)      Spring jms和activemq相关依赖引入

3)      Spring整合activemq配置

4)      定义消息发布者(生产者)

5)      定义消息订阅者(消费者)

6)      Spring mvc配置

7)      实例测试

## 4.      Spring整合JMS和ActiveMq具体实现 ##

### 1)      下载和部署ActiveMq服务器 ###

下载地址：

[http://activemq.apache.org/2016/03/07/apache-activemq-5132-released.html][http_activemq.apache.org_2016_03_07_apache-activemq-5132-released.html]

解压下载文件，假如我保存在D盘根目录,找到目录apache-activemq-5.13.2\\bin\\win64下的activemq.bat,启动activemq服务。

### 2)      Spring jms和activemq相关依赖 ###

<dependency>
               <groupId>org.apache.xbean</groupId>
               <artifactId>xbean-spring</artifactId>
               <version>3.16</version>
           </dependency>
           
           <dependency>
               <groupId>org.apache.activemq</groupId>
               <artifactId>activemq-all</artifactId>
               <version> 5.13.1</version>
         </dependency>
    <dependency>
               <groupId>org.springframework</groupId>
               <artifactId>spring-jms</artifactId>
               <version> 4.2.0.RELEASE</version>
         </dependency>

### 3)      Application-context-jms中配置jms和activemq ###

注意头部信息需要引入jms和activemq

![Center][]

具体配置如下：

<?xml version="1.0"encoding="UTF-8"?>
    <beans xmlns="http://www.springframework.org/schema/beans"
        xmlns:amq="http://activemq.apache.org/schema/core"
        xmlns:jms="http://www.springframework.org/schema/jms"
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xmlns:context="http://www.springframework.org/schema/context"
        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://activemq.apache.org/schema/core
           http://activemq.apache.org/schema/core/activemq-core-5.9.0.xsd
           http://www.springframework.org/schema/jms
           http://www.springframework.org/schema/jms/spring-jms-4.0.xsd">
     
        <description>application-context-activemqconfig</description>
        
        <amq:connectionFactory id="amqConnectionFactory"
           brokerURL="tcp://localhost:61616" userName="admin" password="admin"/>
        <bean id="connectionFactory"
           class="org.springframework.jms.connection.CachingConnectionFactory">
           <constructor-arg ref="amqConnectionFactory"/>
           <property name="sessionCacheSize"value="100" />
        </bean>
        
        
        <bean id="jmsQueueTemplate" class="org.springframework.jms.core.JmsTemplate">
            <constructor-arg ref="connectionFactory" />
           
           <property name="pubSubDomain"value="false" />
        </bean>
        
        <bean id="jmsTopicTemplate" class="org.springframework.jms.core.JmsTemplate">
           <constructor-arg ref="connectionFactory"/>
           
           <property name="pubSubDomain"value="true" />
        </bean>
        
        
        
        <jms:listener-container destination-type="queue"
           container-type="default" connection-factory="connectionFactory"
           acknowledge="auto">
           <jms:listener destination="test.queue"ref="queueReceiver" />
           <jms:listener destination="test.queue"ref="queueReceiver2" />
        </jms:listener-container>
        
        <jms:listener-container destination-type="topic"
           container-type="default" connection-factory="connectionFactory"
           acknowledge="auto">
           <jms:listener destination="test.topic"ref="topicReceiver" />
           <jms:listener destination="test.topic"ref="topicReceiver2" />
        </jms:listener-container>
        
    </beans>

### 4)      定义消息发布者(生产者) ###

#### a.      Queue队列消息发布者 ####

package com.lm.core.service.impl.sender;
     
    importjavax.jms.JMSException;
    importjavax.jms.Message;
    importjavax.jms.Session;
     
    importorg.springframework.beans.factory.annotation.Autowired;
    importorg.springframework.beans.factory.annotation.Qualifier;
    importorg.springframework.jms.core.JmsTemplate;
    importorg.springframework.jms.core.MessageCreator;
    importorg.springframework.stereotype.Component;
     
    @Component
    publicclass QueueSender {
        @Autowired
        @Qualifier("jmsQueueTemplate")
        private JmsTemplate jmsTemplate;// 通过@Qualifier修饰符来注入对应的bean
     
        /**
         * 发送一条消息到指定的队列（目标）
         *
         *@param queueName
         *           队列名称
         *@param message
         *           消息内容
         */
        public void send(String queueName, finalString message) {
                 jmsTemplate.send(queueName, newMessageCreator() {
                           @Override
                           public MessagecreateMessage(Session session) throws JMSException {
                                    returnsession.createTextMessage(message);
                           }
                 });
        }
    }

#### b.      Topic队列信息发布者 ####

packagecom.lm.core.service.impl.sender;
     
    importjavax.jms.JMSException;
    importjavax.jms.Message;
    importjavax.jms.Session;
     
    importorg.springframework.beans.factory.annotation.Autowired;
    importorg.springframework.beans.factory.annotation.Qualifier;
    importorg.springframework.jms.core.JmsTemplate;
    importorg.springframework.jms.core.MessageCreator;
    importorg.springframework.stereotype.Component;
     
    @Component
    publicclass TopicSender {
        @Autowired
        @Qualifier("jmsTopicTemplate")
        private JmsTemplate jmsTemplate;
     
        /**
         * 发送一条消息到指定的队列（目标）
         *
         *@param queueName
         *           队列名称
         *@param message
         *           消息内容
         */
        public void send(String topicName, finalString message) {
                 jmsTemplate.send(topicName, newMessageCreator() {
                           @Override
                           public MessagecreateMessage(Session session) throws JMSException {
                                    returnsession.createTextMessage(message);
                           }
                 });
        }
    }

### 5)      定义消息订阅者(消费者) ###

#### a)        Queue队列消息接受者1 ####

#### b)        Queue队列消息接受者2 ####

packagecom.lm.core.service.impl.receiver;
     
    importjavax.jms.JMSException;
    importjavax.jms.Message;
    importjavax.jms.MessageListener;
    importjavax.jms.TextMessage;
     
    importorg.springframework.stereotype.Component;
     
    @Component
    publicclass QueueReceiver2 implements MessageListener {
             @Override
             public void onMessage(Message message){
                       try {
                                System.out.println("QueueReceiver2接收到消息:"
                                                   +((TextMessage) message).getText());
                       } catch (JMSException e) {
                                e.printStackTrace();
                       }
             }
    }
    c)        Topic队列消息接受者1
    packagecom.lm.core.service.impl.receiver;
     
    importjavax.jms.JMSException;
    importjavax.jms.Message;
    importjavax.jms.MessageListener;
    importjavax.jms.TextMessage;
     
    importorg.springframework.stereotype.Component;
     
    @Component
    public classTopicReceiver implements MessageListener {
             @Override
             public void onMessage(Message message){
                       try {
                                System.out.println("TopicReceiver1接收到消息:"
                                                   +((TextMessage) message).getText());
                       } catch (JMSException e) {
                                e.printStackTrace();
                       }
             }
    }

#### d)        Topic队列消息接受者2 ####

### 6)      Spring mvc配置 ###

注意：需要通过mvc:resources配置静态资源，否则找不到相关依赖的资源

![Center 1][]

<?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:mvc="http://www.springframework.org/schema/mvc"
        xsi:schemaLocation="http://www.springframework.org/schema/beans 
                            http://www.springframework.org/schema/beans/spring-beans-3.1.xsd 
                           http://www.springframework.org/schema/context 
                           http://www.springframework.org/schema/context/spring-context-3.1.xsd 
                            http://www.springframework.org/schema/mvc 
                           http://www.springframework.org/schema/mvc/spring-mvc-4.0.xsd">
        <description>spring-mvc config</description>
     
        
        <mvc:annotation-driven />
     
        
        <context:component-scan base-package="com.lm.web" />
     
        
        <bean id="mappingJacksonHttpMessageConverter"
           class="org.springframework.http.converter.json.MappingJackson2HttpMessageConverter">
           <property name="supportedMediaTypes">
               <list>
                  <value>text/html;charset=UTF-8</value>
               </list>
           </property>
        </bean>
        
        <bean
           class="org.springframework.web.servlet.mvc.annotation.AnnotationMethodHandlerAdapter">
           <property name="messageConverters">
               <list>
                  <ref bean="mappingJacksonHttpMessageConverter"/> 
               </list>
           </property>
        </bean>
        
        <mvc:resources location="/resources/" mapping="/resources/**" />
        
        <bean
           class="org.springframework.web.servlet.view.InternalResourceViewResolver">
           
           <property name="prefix"value="/WEB-INF/view/" />
           <property name="suffix"value=".jsp" />
        </bean>
     
        
        <bean id="multipartResolver"
           class="org.springframework.web.multipart.commons.CommonsMultipartResolver">
           
           <property name="defaultEncoding"value="utf-8" />
           
           <property name="maxUploadSize"value="10485760000" />
           
           <property name="maxInMemorySize"value="40960" />
        </bean>
    </beans>

### 7)      实例测试 ###

#### a)      Queue响应信息 ####

从日志可以看出，Queue模式是单点响应消息

![Center 2][]

![Center 3][]

#### b)      Topic响应信息 ####

从日志可以看出，topic模式时，不管你有多少订阅者，同时响应消息

![Center 4][]

# 三.  多服务器异步访问(远程访问) #

上一节我们实现了单web服务器响应异步通信。本节将实现多web服务器实现异步通信。

## 1.      实现流程 ##

1)      引入jms和activemq相关的依赖

2)      Spring 配置jms和activemq

3)      定义消息订阅者(消费者)

4)      实例测试

## 2.      具体实现 ##

### 1)      引入jms和activemq相关的依赖 ###

类似上一节依赖

### 2)      Spring 配置jms和activemq ###

类似上一节配置，只配置订阅者,如下：

<?xml version="1.0"encoding="UTF-8"?>
    <beans xmlns="http://www.springframework.org/schema/beans"
        xmlns:amq="http://activemq.apache.org/schema/core"
        xmlns:jms="http://www.springframework.org/schema/jms"
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xmlns:context="http://www.springframework.org/schema/context"
        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://activemq.apache.org/schema/core
           http://activemq.apache.org/schema/core/activemq-core-5.9.0.xsd
           http://www.springframework.org/schema/jms
            http://www.springframework.org/schema/jms/spring-jms-4.0.xsd">
     
        <description>application-context-activemqconfig</description>
     
        
        <amq:connectionFactory id="amqConnectionFactory"
           brokerURL="tcp://localhost:61616" userName="admin" password="admin"/>
          
        <bean id="connectionFactory"
           class="org.springframework.jms.connection.CachingConnectionFactory">
           <constructor-arg ref="amqConnectionFactory"/>
           <property name="sessionCacheSize"value="100" />
        </bean>
        
        
        
        
        <jms:listener-container destination-type="topic"
           container-type="default" connection-factory="connectionFactory"
           acknowledge="auto">
           <jms:listener destination="test.topic"ref="topicReceiver" />
           <jms:listener destination="test.topic"ref="topicReceiver2" />
        </jms:listener-container>
    </beans>

## 3)      定义消息订阅者(消费者) ##

### a.      Topic模式订阅者1 ###

package com.lm.core.service.receiver;
     
    importjavax.jms.JMSException;
    importjavax.jms.Message;
    importjavax.jms.MessageListener;
    importjavax.jms.TextMessage;
     
    importorg.springframework.stereotype.Component;
     
    @Component
    public classTopicReceiver implements MessageListener {
           @Override
           public void onMessage(Message message) {
                     try {
                              System.out.println("TopicReceiver1client接收到消息:"
                                                 +((TextMessage) message).getText());
                     } catch (JMSException e) {
                              e.printStackTrace();
                     }
           }
    }

### b.      Topic模式订阅者2 ###

packagecom.lm.core.service.receiver;
     
    importjavax.jms.JMSException;
    importjavax.jms.Message;
    importjavax.jms.MessageListener;
    importjavax.jms.TextMessage;
     
    importorg.springframework.stereotype.Component;
     
    @Component
    publicclass TopicReceiver2 implements MessageListener {
             @Override
             public void onMessage(Message message){
                      try {
                               System.out.println("TopicReceiver2client:接收到消息:"
                                                  +((TextMessage) message).getText());
                      } catch (JMSException e) {
                               e.printStackTrace();
                      }
             }
    }

## 4)      实例测试 ##

启动activemq服务,tomcat启动server和client

从日志可以看出发送消息时，server和client响应到消息。

![Center 3][]

![Center 5][]

参考文献：

**[Spring和ActiveMQ集成实现队列消息以及PUB/SUB模型][Spring_ActiveMQ_PUB_SUB]**

**代码地址：[http://download.csdn.net/detail/a123demi/9498084][http_download.csdn.net_detail_a123demi_9498084]**

**Git地址：[https://github.com/a123demi/spirng-mybatis-remote][https_github.com_a123demi_spirng-mybatis-remote]**

[http_activemq.apache.org]: http://activemq.apache.org/
[http_activemq.apache.org_2016_03_07_apache-activemq-5132-released.html]: http://activemq.apache.org/2016/03/07/apache-activemq-5132-released.html
[Center]: /images/20220724/c082407bd1f647a7b34f4274f6871b99.png
[Center 1]: /images/20220724/16f306ff05514dbba25a5e127afc061d.png
[Center 2]: /images/20220724/e28c8b4c69314d0ab65f9dd5bb1864e1.png
[Center 3]: /images/20220724/61369941cbbc492eaff2cfb1ba97826b.png
[Center 4]: /images/20220724/4e9009c4811147ec85fe9406c54bf70e.png
[Center 5]: /images/20220724/17a0051d93c24fa0953261b72d013763.png
[Spring_ActiveMQ_PUB_SUB]: http://www.xdemo.org/spring-activemq-pub-sub/
[http_download.csdn.net_detail_a123demi_9498084]: http://download.csdn.net/detail/a123demi/9498084
[https_github.com_a123demi_spirng-mybatis-remote]: https://github.com/a123demi/spirng-mybatis-remote