ActiveMQ——7.Spring-jms的配置

太过爱你忘了你带给我的痛 2021-09-26 09:26 289阅读 0赞

## 概述 ##

--------------------

如何使用spring-jms来简化jms客户端的开发？

这篇文章主要记录如何配置以便以后复用。

## producer端 ##

--------------------

producer端负责发送，这里使用JmsTemplate。

## spring配置 ##

<?xml version="1.0" encoding="UTF-8"?>
    <beans xmlns="http://www.springframework.org/schema/beans"
        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">
            
        <bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
            <property name="brokerURL" value="tcp://localhost:61616" />
        </bean>
        
        
        <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
            
            <property name="connectionFactory" ref="connectionFactory" />
            <property name="defaultDestinationName" value="jms-config" />
        </bean>
    </beans>

JmsTemplate默认将jms-config解析为Queue类型的Destination。如果需要将其解析为Topic类型，需要为jmsTemplate指定属性 pubSubDomain=true ，配置如下：

**测试类**

package cn.sinobest.asj.producer.springsupport.jt;
    import javax.annotation.Resource;
    
    import org.junit.Test;
    import org.junit.runner.RunWith;
    import org.springframework.jms.core.JmsTemplate;
    import org.springframework.test.context.ContextConfiguration;
    import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
    
    @RunWith(SpringJUnit4ClassRunner.class) // 配置spring组件运行时
    @ContextConfiguration(locations="classpath:spring-jms-demo.xml") // 配置文件
    public class JmsTemplateSendWithContextTest {
        @Resource(name = "jmsTemplate")
        private JmsTemplate jt;
        
        @Test
        public void testSendToDefaultDestination() {
            String message = "you can config JmsTemplate in xml, then use it for send.";
            jt.convertAndSend(message);
        }
    }

展示这个测试类，是想告诉大家使用Spring+JUnit4的注解编写单元测试，可以非常方便的加载Spring配置并初始化bean资源；使用Resource注解可以获取bean资源。

## consumer端 ##

--------------------

consumer端负责接收，接收有同步、异步两种方式 。

## 同步接收 ##

这里使用JmsTemplate进行同步接收。上面已经给过了使用JmsTemplate发送的配置， 接收和发送的配置能有什么区别呢？

如果我们不希望客户端一直阻塞等待消息，那么需要关心receiveTimeout属性，单位毫秒。如果超过了这个时间，还没有接收到消息，就返回null。

## 异步接收 ##

异步接收是基于监听器的接收，传统的配置方式是配置一个ListenerContainer的bean，在这个bean里维护listener。还有一种精简的配置方案，可以在一个container中放置多个listener。

### 传统的配置 ###

<?xml version="1.0" encoding="UTF-8"?>
    <beans xmlns="http://www.springframework.org/schema/beans"
        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">
            
        <bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
            <property name="brokerURL" value="tcp://localhost:61616" />
        </bean>
        
        <bean id="messageListener" class="cn.sinobest.asj.consumer.springsupport.async.SimpleMessageListener" />
        <bean id="messageContainer"
            class="org.springframework.jms.listener.DefaultMessageListenerContainer">
            <property name="connectionFactory" ref="connectionFactory" />
            <property name="destinationName" value="jms-config" />
            <property name="messageListener" ref="messageListener" />
        </bean>
    </beans>

其中，SimpleMessageListener是接口javax.jms.MessageListener的实现类。

DefaultMessageListenerContainer负责将messageListener注册到connectionFactory的destination，一旦destination中有消息，就会将消息推送给 messageListener。

DefaultMessageListenerContainer有很多特性的配置，下面择要介绍：

#### 1.Destination及类型 ####

使用下面的API，可以设置Destination

<table style="margin:0px; padding:0px; border-collapse:collapse; border-spacing:0px; max-width:850px; border:1px solid silver; word-break:break-word; width:847px"> 
 <tbody style="margin:0px; padding:0px"> 
  <tr style="margin:0px; padding:0px"> 
   <td style="margin:0px; padding:3px; border:1px solid silver; border-collapse:collapse"> 
    <ul style="margin:0px 0px 1em 40px; padding:0px"> 
     <li style="margin:0px; padding:0px; list-style-type:disc">public void setPubSubDomain(boolean pubSubDomain)<br style="margin:0px; padding:0px"> 设置destination的类型，true-Topics，false-Queues；默认为false。</li> 
     <li style="margin:0px; padding:0px; list-style-type:disc">public void setDestinationName(String destinationName)<br style="margin:0px; padding:0px"> 设置destination的name，结合pubSubDomain使用，根据destinationName解析为具体的Destination。</li> 
     <li style="margin:0px; padding:0px; list-style-type:disc">public void setDestination(Destination destination)<br style="margin:0px; padding:0px"> 设置destination。</li> 
    </ul> </td> 
  </tr> 
 </tbody> 
</table>

#### 2.多线程 ####

一个DMLC 的实例，只能管理一个 MessageListener实例，但是可以使用下面的方法设置多线程：

如果没有使用事务，多线程可以显著提高监听器的接收速度。

#### 3.确认模式 ####

下面的API用来设置确认模式：

<table style="margin:0px; padding:0px; border-collapse:collapse; border-spacing:0px; max-width:850px; border:1px solid silver; word-break:break-word; width:847px"> 
 <tbody style="margin:0px; padding:0px"> 
  <tr style="margin:0px; padding:0px"> 
   <td style="margin:0px; padding:3px; border:1px solid silver; border-collapse:collapse"> 
    <ul style="margin:0px 0px 1em 40px; padding:0px"> 
     <li style="margin:0px; padding:0px; list-style-type:disc">public void setSessionAcknowledgeMode(int sessionAcknowledgeMode)<br style="margin:0px; padding:0px"> 设置确认模式，sessionAcknowledgeMode可以取javax.jms.Session.AUTO_ACKNOWLEDGE（默认），javax.jms.Session.CLIENT_ACKNOWLEDGE，javax.jms.Session.DUPS_OK_ACKNOWLEDGE。</li> 
     <li style="margin:0px; padding:0px; list-style-type:disc">public void setSessionAcknowledgeModeName(String constantName)<br style="margin:0px; padding:0px"> 通过名字来设置确认模式，默认为"AUTO_ACKNOWLEDGE"。</li> 
    </ul> </td> 
  </tr> 
 </tbody> 
</table>

#### 4.事务 ####

下面的API，用来设置事务：

更多DefaultMessageListenerContainer的相关配置，参考其API。

### 精简的配置 ###

称之为精简版的配置，因为一个容器可以配置多个监听器。

<?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:amq="http://activemq.apache.org/schema/core"
           xmlns:jms="http://www.springframework.org/schema/jms"
           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-4.1.xsd
            http://www.springframework.org/schema/context
            http://www.springframework.org/schema/context/spring-context-4.1.xsd
            http://www.springframework.org/schema/mvc
            http://www.springframework.org/schema/mvc/spring-mvc-4.1.xsd
            http://www.springframework.org/schema/jms
            http://www.springframework.org/schema/jms/spring-jms-4.1.xsd
            http://activemq.apache.org/schema/core
            http://activemq.apache.org/schema/core/activemq-core-5.12.1.xsd">
            
        <bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
            <property name="brokerURL" value="tcp://localhost:61616" />
        </bean>
            
        <jms:listener-container destination-type="topic" container-type="default" connection-factory="connectionFactory" acknowledge="auto">    
          	
            <jms:listener destination="test.topic" ref="topicReceiver1"/>    
            <jms:listener destination="test.topic" ref="topicReceiver2"/>    
        </jms:listener-container>     
    </beans>

注意：精简版的 schema 和传统版不一样，详细的精简版可以参考[Spring整个ActiveMQ（精简版配置）][Spring_ActiveMQ]

listener-container作为容器，可以有多个listener子元素，每个 listener代表一个监听器。

#### 1.Destination类型 ####

listener-container有 destination-type属性，可以取值\["queue", "topic"\]，默认为"queue"，它决定了 listener将destination解析为Queue还是Topic类型。

#### 2.pojo监听器 ####

listener会对消息进行转换，所以ref的目标bean是一个pojo，method是这个pojo的方法的名字——这个方法的参数要和Message中的数据类型兼容。下面给出我们使用的 topicReceiver1：

package com.gwd.mq.producer.topic;    
        
    import javax.jms.JMSException;    
    import javax.jms.Message;    
    import javax.jms.MessageListener;    
    import javax.jms.TextMessage;    
    import org.springframework.stereotype.Component;    
         
    @Component    
    public class TopicReceiver1 implements MessageListener{    
        
        public void onMessage(Message message) {    
            try {    
                System.out.println("TopicReceiver1接收到消息:"+((TextMessage)message).getText());    
            } catch (JMSException e) {    
                e.printStackTrace();    
            }    
        }    
    }

#### 3.多线程 ####

为listener-container元素设置concurrency属性，可以指定 线程数的下限和上限。这一点和 DefaultMessageListenerContainer的相同 。

#### 4.确认模式和事务4 ####

listener-container的acknowledge属性，可以指定确认模式或者事务，取值范围\["auto", "client", "dups-ok", "transacted"\]；默认为"auto"，事务使用" transacted "。

更多精简配置相关的参数，参考 [spring-jms-4.2.xsd][]

## 可信任的包 ##

ObjectMessage的使用机制是不安全的，ActiveMQ自5.12.2和5.13.0之后，强制consumer端声明一份可信任的包列表，只有当ObjectMessage中的Object在可信任包内，才能被提取出来。

添加可信任的一般常用两种，一种按自己的需求添加，另一种是添加全部可信任

//按需添加信任代码版设置
    ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
    factory.setTrustedPackages(new ArrayList(Arrays.asList("org.apache.activemq.test,org.apache.camel.test".split(","))));

//完全信任的代码版设置
    ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
    factory.setTrustAllPackages(true);

配置版分别对应如下：

<bean id="connectionFactory" class="org.apache.activemq.spring.ActiveMQConnectionFactory">
        <property name="brokerURL" value="tcp://localhost:61616"/>
        <property name="trustedPackages">
            <list>
                <value>org.apache.activemq.test</value>
                <value>org.apache.camel.test</value>
            </list>
        </property>
    </bean>
    <bean id="jmsConfig" class="org.apache.camel.component.jms.JmsConfiguration">
        <property name="connectionFactory" ref="connectionFactory"/>
    </bean>
    <bean id="activemq" class="org.apache.activemq.camel.component.ActiveMQComponent">
        <property name="configuration" ref="jmsConfig"/>
    </bean>

关于ObjectMessage安全性的说明，参考 [官方文档][Link 1]

## ConnectionFactory的bean配置 ##

一般有如下几种：

## 1.ActiveMQConnectionFactory ##

在前面的配置中，我们已经接触了这个配置：

## 2.SingleConnectionFactory ##

使用org.springframework.jms.connection.SingleConnectionFactory对ActiveMQConnectionFactory进行包装，建议用于测试或者单机的环境。

## 3.PooledConnectionFactory ##

使用org.apache.activemq.pool.PooledConnectionFactory对ActiveMQConnectionFactory进行包装。

**方式的选择：**

1.如果只需要一个connection选择 ActiveMQConnectionFactory，如果需要多个connection选择PooledConnectionFactory

2.By default the pooled version will Idle out old connections after some time of non-use. The ActiveMQ connection created from its factory is valid until you call close() or it is no longer referenced。（摘自stackoverflow，大致意思是PooledConnectionFactory会释放长连接，ActiveMQConnectionFactory不会释放直到你关闭它或者不再引用）

3.SingleConnectionFactory一般用于测试或者单机环境

[Spring_ActiveMQ]: http://blog.csdn.net/gwd1154978352/article/details/78756611
[spring-jms-4.2.xsd]: http://www.springframework.org/schema/jms/spring-jms-4.2.xsd
[Link 1]: http://activemq.apache.org/objectmessage.html