AMQ学习笔记 - 13. Spring-jms的配置

红太狼 2022-04-10 12:43 256阅读 0赞

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

这篇文章主要记录如何配置以便以后复用，而非原理的讲解，有些内容我 没有掌握原理。

## producer端 ##

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

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

## spring配置 ##

![复制代码][copycode.gif]

1 <?xml version="1.0" encoding="UTF-8"?>
     2 <beans xmlns="http://www.springframework.org/schema/beans"
     3     xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
     4     xsi:schemaLocation="http://www.springframework.org/schema/beans
     5         http://www.springframework.org/schema/beans/spring-beans-4.2.xsd">
     6         
     7     <bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
     8         <property name="brokerURL" value="tcp://localhost:61616" />
     9     </bean>
    10     
    11     
    12     <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
    13         
    14         <property name="connectionFactory" ref="connectionFactory" />
    15         <property name="defaultDestinationName" value="jms-config" />
    16     </bean>
    17 </beans>

![复制代码][copycode.gif]

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

![复制代码][copycode.gif]

1 <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
    2     
    3     <property name="connectionFactory" ref="connectionFactory" />
    4     <property name="pubSubDomain" value="true" />
    5     <property name="defaultDestinationName" value="jms-config" />
    6 </bean>

![复制代码][copycode.gif]

## 测试类 ##

![复制代码][copycode.gif]

1 package cn.sinobest.asj.producer.springsupport.jt;
     2 import javax.annotation.Resource;
     3 
     4 import org.junit.Test;
     5 import org.junit.runner.RunWith;
     6 import org.springframework.jms.core.JmsTemplate;
     7 import org.springframework.test.context.ContextConfiguration;
     8 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
     9 
    10 @RunWith(SpringJUnit4ClassRunner.class) // 配置spring组件运行时
    11 @ContextConfiguration("/spring-jms-demo.xml") // 配置文件
    12 public class JmsTemplateSendWithContextTest {
    13     @Resource(name = "jmsTemplate")
    14     private JmsTemplate jt;
    15     
    16     @Test
    17     public void testSendToDefaultDestination() {
    18         String message = "you can config JmsTemplate in xml, then use it for send.";
    19         jt.convertAndSend(message);
    20     }
    21 }

![复制代码][copycode.gif]

展示这个测试类，是想告诉大家使用Spring+JUnit4的注解编写单元测试，可以非常方便的加载Spring配置并初始化bean资源；使用Resource注解可以获取bean资源。如何使用JmsTemplate来发送消息，反而不是重点，因为在[08. Spring-JmsTemplate之发送][08. Spring-JmsTemplate]中，已经详细介绍了相关的API。

## consumer端 ##

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

consumer端负责接收，接收有同步、异步两种方式，在[03. 消息的接收方式][03.]中有所介绍。

## 同步接收 ##

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

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

![复制代码][copycode.gif]

1 <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
    2     
    3     <property name="connectionFactory" ref="connectionFactory" />
    4     <property name="defaultDestinationName" value="jms-config" />
    5     <property name="receiveTimeout" value="3000" />
    6 </bean>

![复制代码][copycode.gif]

## 异步接收 ##

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

### 传统的配置 ###

![复制代码][copycode.gif]

1 <?xml version="1.0" encoding="UTF-8"?>
     2 <beans xmlns="http://www.springframework.org/schema/beans"
     3     xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
     4     xsi:schemaLocation="http://www.springframework.org/schema/beans
     5         http://www.springframework.org/schema/beans/spring-beans-4.2.xsd">
     6         
     7     <bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
     8         <property name="brokerURL" value="tcp://localhost:61616" />
     9     </bean>
    10     
    11     <bean id="messageListener" class="cn.sinobest.asj.consumer.springsupport.async.SimpleMessageListener" />
    12     <bean id="messageContainer"
    13         class="org.springframework.jms.listener.DefaultMessageListenerContainer">
    14         <property name="connectionFactory" ref="connectionFactory" />
    15         <property name="destinationName" value="jms-config" />
    16         <property name="messageListener" ref="messageListener" />
    17     </bean>
    18 </beans>

![复制代码][copycode.gif]

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

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

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

1.Destination及类型

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

<table> 
 <tbody> 
  <tr> 
   <td> 
    <ul> 
     <li>public void setPubSubDomain(boolean pubSubDomain)<br> 设置destination的类型，true-Topics，false-Queues；默认为false。</li> 
     <li>public void setDestinationName(String destinationName)<br> 设置destination的name，结合pubSubDomain使用，根据destinationName解析为具体的Destination。</li> 
     <li>public void setDestination(Destination destination)<br> 设置destination。</li> 
    </ul></td> 
  </tr> 
 </tbody> 
</table>

2.多线程

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

<table> 
 <tbody> 
  <tr> 
   <td> 
    <ul> 
     <li>public void setConcurrency(String concurrency)<br> 通过"lower-upper"格式的字符串，设置线程数的下限和上限，如"5-10"；或者仅设置上限，下限默认为1，如"10"。</li> 
    </ul></td> 
  </tr> 
 </tbody> 
</table>

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

3.确认模式

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

<table> 
 <tbody> 
  <tr> 
   <td> 
    <ul> 
     <li>public void setSessionAcknowledgeMode(int sessionAcknowledgeMode)<br> 设置确认模式，sessionAcknowledgeMode可以取javax.jms.Session.AUTO_ACKNOWLEDGE（默认），javax.jms.Session.CLIENT_ACKNOWLEDGE，javax.jms.Session.DUPS_OK_ACKNOWLEDGE。</li> 
     <li>public void setSessionAcknowledgeModeName(String constantName)<br> 通过名字来设置确认模式，默认为"AUTO_ACKNOWLEDGE"。</li> 
    </ul></td> 
  </tr> 
 </tbody> 
</table>

4.事务

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

<table> 
 <tbody> 
  <tr> 
   <td> 
    <ul> 
     <li> <p>public void setSessionTransacted(boolean sessionTransacted)</p> <p>设置是否使用事务，默认为false。</p> </li> 
    </ul></td> 
  </tr> 
 </tbody> 
</table>

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

### 精简的配置 ###

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

![复制代码][copycode.gif]

1 <?xml version="1.0" encoding="UTF-8"?>
     2 <beans xmlns="http://www.springframework.org/schema/beans"
     3     xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
     4     xsi:schemaLocation="http://www.springframework.org/schema/beans
     5         http://www.springframework.org/schema/beans/spring-beans-4.2.xsd">
     6         
     7     <bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
     8         <property name="brokerURL" value="tcp://localhost:61616" />
     9     </bean>
    10     
    11     <bean id="messageListener" class="cn.sinobest.asj.consumer.springsupport.async.SimpleMessageListener" />
    12     <bean id="messageContainer"
    13         class="org.springframework.jms.listener.DefaultMessageListenerContainer">
    14         <property name="connectionFactory" ref="connectionFactory" />
    15         <property name="destinationName" value="jms-config" />
    16         <property name="messageListener" ref="messageListener" />
    17     </bean>
    18 </beans>

![复制代码][copycode.gif]

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中的数据类型兼容。下面给出我们使用的PojoListener：

![复制代码][copycode.gif]

1 package cn.sinobest.asj.consumer.springsupport.async;
     2 
     3 /**
     4  * 一个pojo作为listener，接收经过转换后的消息.
     5  * @author lijinlong
     6  *
     7  */
     8 public class PojoListener {
     9     public void passMeMessage(String message) {
    10         System.out.println("从queue收到消息：" + message);
    11     }
    12 }

![复制代码][copycode.gif]

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在可信任包内，才能被提取出来。

你可以这样配置可信任包：

![复制代码][copycode.gif]

1 <bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
     2     <property name="brokerURL" value="tcp://localhost:61616" />
     3     <property name="trustedPackages">
     4         <list>
     5             
     6             <value>java.lang</value>
     7             <value>java.sql</value>
     8             <value>cn.sinobest</value>
     9         </list>
    10     </property>
    11 </bean>

![复制代码][copycode.gif]

关于ObjectMessage安全性的说明，参考[http://activemq.apache.org/objectmessage.html][http_activemq.apache.org_objectmessage.html]

## ConnectionFactory的bean配置 ##

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

我在阅读网友的博文的过程中，见过几种配置的方式，但我并不了解它们之间的优劣区别。

## 1.ActiveMQConnectionFactory ##

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

1 <bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
    2     <property name="brokerURL" value="tcp://localhost:61616" />
    3 </bean>

## 2.SingleConnectionFactory ##

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

![复制代码][copycode.gif]

1 <bean id="connectionFactory"
    2     class="org.springframework.jms.connection.SingleConnectionFactory">
    3     <property name="targetConnectionFactory">
    4         <bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
    5             <property name="brokerURL" value="tcp://localhost:61616" />
    6         </bean>
    7     </property>
    8 </bean>

![复制代码][copycode.gif]

## 3.PooledConnectionFactory ##

使用org.apache.activemq.pool.PooledConnectionFactory对ActiveMQConnectionFactory进行包装，暂不知有什么优化。

![复制代码][copycode.gif]

1 <bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory">
    2     <property name="connectionFactory">
    3         <bean class="org.apache.activemq.ActiveMQConnectionFactory">
    4             <property name="brokerURL" value="tcp://localhost:61616" />
    5         </bean>
    6     </property>
    7 </bean>

![复制代码][copycode.gif]

## 分布式事务 ##

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

首先引入数据源的概念，数据源能提供数据或者能存储数据，一般是两个特点都具备。如数据库、消息队列，都是数据源。

如果你要使用多个数据源，那就要考虑分布式事务。比如从broker-A中接收消息，发送到broker-B。如果在发送的过程中发生错误，接收的消息就不应该确认，否则会被broker-A移除，造成消息的丢失。或者你想从broker-A中接收消息，写入数据库，也有同样的问题。而分布式事务要处理的问题，就是涉及到多个数据源的事务问题，保证涉及多个数据源的操作要么同时成功，要么同时失败。

关于分布式事务的讨论，可选的参考[XA事务处理][XA]，下面针对两种情景，讲一下配置。

## 1.broker-A到broker-B ##

从broker-A中接收消息，发送到broker-B中。在这样的情景中可能会出现：应用系统将数据发送到broker-B，发送失败的数据先存储到broker-A，然后由定时任务从broker-A中获取数据，发送到broker-B。

在这里我们使用同步接收的方式，接收和发送都是基于JmsTemplate的。

Spring提供了org.springframework.transaction.jta.JtaTransactionManager，但是依赖于底层的应用服务器支持JTA全局事务。在这里我们没有使用这样的服务器，而是用Atomikos框架。

事务的配置：

![复制代码][copycode.gif]

1 <?xml version="1.0" encoding="UTF-8"?>
     2 <beans xmlns="http://www.springframework.org/schema/beans"
     3     xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
     4     xmlns:tx="http://www.springframework.org/schema/tx"
     5     xsi:schemaLocation="http://www.springframework.org/schema/beans
     6         http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
     7         http://www.springframework.org/schema/tx
     8         http://www.springframework.org/schema/tx/spring-tx-3.0.xsd">
     9     
    10     <bean id="jtaTransactionManager"
    11         class="org.springframework.transaction.jta.JtaTransactionManager">
    12         <property name="transactionManager">
    13             <bean class="com.atomikos.icatch.jta.UserTransactionManager"
    14                 init-method="init" destroy-method="close">
    15                 <property name="forceShutdown" value="false" />
    16             </bean>
    17         </property>
    18         <property name="userTransaction">
    19             <bean class="com.atomikos.icatch.jta.UserTransactionImp">
    20                 <property name="transactionTimeout" value="300" />
    21             </bean>
    22         </property>
    23     </bean>
    24     
    25     <tx:annotation-driven transaction-manager="jtaTransactionManager" />
    26     
    27     
    28     <bean id="sourceConnFactory" class="com.atomikos.jms.AtomikosConnectionFactoryBean">
    29         <property name="uniqueResourceName" value="broker/source" />
    30         <property name="xaConnectionFactory">
    31             <bean class="org.apache.activemq.ActiveMQXAConnectionFactory">
    32                 <property name="brokerURL" value="tcp://localhost:57015" />
    33             </bean>
    34         </property>
    35         <property name="maxPoolSize" value="10" />
    36     </bean>
    37     
    38     <bean id="targetConnFactory" class="com.atomikos.jms.AtomikosConnectionFactoryBean">
    39         <property name="uniqueResourceName" value="broker/target" />
    40         <property name="xaConnectionFactory">
    41             <bean class="org.apache.activemq.ActiveMQXAConnectionFactory">
    42                 <property name="brokerURL" value="tcp://localhost:61616" />
    43             </bean>
    44         </property>
    45         <property name="maxPoolSize" value="10" />
    46     </bean>
    47     
    48     <bean id="sourceJmsTemplate" class="org.springframework.jms.core.JmsTemplate">
    49         <property name="connectionFactory" ref="sourceConnFactory" />
    50         <property name="defaultDestinationName" value="asj.log" />
    51         <property name="receiveTimeout" value="3000" />
    52         <property name="sessionTransacted" value="true" />
    53     </bean>
    54     
    55     <bean id="targetJmsTemplate" class="org.springframework.jms.core.JmsTemplate">
    56         <property name="connectionFactory" ref="targetConnFactory" />
    57         <property name="defaultDestinationName" value="asj.log" />
    58         <property name="sessionTransacted" value="true" />
    59     </bean>
    60 </beans>

![复制代码][copycode.gif]

服务类获取sourceJmsTemplate、targetJmsTemplate，在业务方法中使用前者接收消息，使用后者发送消息。业务代码要注解为org.springframework.transaction.annotation.Transactional：

![复制代码][copycode.gif]

1 @Transactional
     2 public boolean transport() throws JmsException{
     3     Message remsg = souJT.receive();
     4     if (remsg != null) {
     5         final ObjectMessage omsg = (ObjectMessage) remsg;
     6         
     7         tarJT.send(new MessageCreator() {
     8             public Message createMessage(Session session)
     9                     throws JMSException {
    10                 Message result = session.createObjectMessage(omsg
    11                         .getObject());
    12                 return result;
    13             }
    14         });
    15     }
    16     return remsg != null;
    17 }

![复制代码][copycode.gif]

实际上，我虽然在项目里用了上面的配置，但是对其原理却不了解。另外有2个问题值得一说：

1.  定时任务和事务的冲突  
    服务类是事务的，如果它同时作为定时任务，会有问题；后来把定时任务独立了出来，定时调用服务类的业务方法。
2.  事务和循环的冲突  
    本来计划在事务中使用循环，以souJT接收到的消息为null作为结束条件，每接收一条就发送一条。结果即使有更多的消息，也只执行一次，是不是由Atomikos造成的也无从知晓。后来就把循环放在了定时任务里，在循环体调用业务方法，根据返回的结果来判断是否结束循环。

## 2.broker-B到数据库 ##

从broker-B中接收消息，写入数据库。这一次没有使用Atomikos，也实现了当写库失败时回滚消息的事务效果；而前一个小节broker-A到broker-B，如果不使用Atomikos就无法达到事务效果。或许因为它使用的是同步接收，而接下来我们使用的是异步接收。接收的方式是否有对分布式事务有关系，结论还不能确定。

在这里，我们甚至没有再使用spring的JtaTransactionManager。

![复制代码][copycode.gif]

当然，我们必须声明listener-container的acknowledge属性为"transacted"，以开启事务。在ConverteredLogListener的onLog，调用服务组件将数据入库，只要写入失败时抛出的异常能抵达onLog方法，事务就会回滚，所以请确保异常能抛出来。

[copycode.gif]: /images/20220330/fe05128d7e03464bb8b09ac617de5891.png
[08. Spring-JmsTemplate]: http://www.cnblogs.com/ywjy/p/5434013.html
[03.]: http://www.cnblogs.com/ywjy/p/5433813.html
[spring-jms-4.2.xsd]: http://www.springframework.org/schema/jms/spring-jms-4.2.xsd
[http_activemq.apache.org_objectmessage.html]: http://activemq.apache.org/objectmessage.html
[XA]: http://www.infoq.com/cn/articles/xa-transactions-handle