概述


如何使用spring-jms来简化jms客户端的开发?
这篇文章主要记录如何配置以便以后复用,而非原理的讲解,有些内容我 没有掌握原理。

producer端


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

spring配置

 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     <!-- create template for send message --> 
12     <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate"> 
13         <!-- bind the connection factory --> 
14         <property name="connectionFactory" ref="connectionFactory" /> 
15         <property name="defaultDestinationName" value="jms-config" /> 
16     </bean> 
17 </beans>
JmsTemplate默认将jms-config解析为Queue类型的Destination。如果需要将其解析为Topic类型,需要为jmsTemplate指定属性 pubSubDomain=true,配置如下:
1 <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate"> 
2     <!-- bind the connection factory --> 
3     <property name="connectionFactory" ref="connectionFactory" /> 
4     <property name="pubSubDomain" value="true" /> 
5     <property name="defaultDestinationName" value="jms-config" /> 
6 </bean>

测试类

 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 }

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

consumer端


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

同步接收

这里使用JmsTemplate进行同步接收。上面已经给过了使用JmsTemplate发送的配置, 接收和发送的配置能有什么区别呢?
如果我们不希望客户端一直阻塞等待消息,那么需要关心receiveTimeout属性,单位毫秒。如果超过了这个时间,还没有接收到消息,就返回null。
1 <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate"> 
2     <!-- bind the connection factory --> 
3     <property name="connectionFactory" ref="connectionFactory" /> 
4     <property name="defaultDestinationName" value="jms-config" /> 
5     <property name="receiveTimeout" value="3000" /> 
6 </bean>

异步接收

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

传统的配置

 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>

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

DefaultMessageListenerContainer负责将messageListener注册到connectionFactory的destination,一旦destination中有消息,就会将消息推送给 messageListener。
DefaultMessageListenerContainer有很多特性的配置,下面择要介绍:

1.Destination及类型

使用下面的API,可以设置Destination
  • public void setPubSubDomain(boolean pubSubDomain)
    设置destination的类型,true-Topics,false-Queues;默认为false。
  • public void setDestinationName(String destinationName)
    设置destination的name,结合pubSubDomain使用,根据destinationName解析为具体的Destination。
  • public void setDestination(Destination destination)
    设置destination。

2.多线程

一个DMLC 的实例,只能管理一个 MessageListener实例,但是可以使用下面的方法设置多线程:
  • public void setConcurrency(String concurrency)
    通过"lower-upper"格式的字符串,设置线程数的下限和上限,如"5-10";或者仅设置上限,下限默认为1,如"10"。
如果没有使用事务,多线程可以显著提高监听器的接收速度。

3.确认模式

下面的API用来设置确认模式:
  • public void setSessionAcknowledgeMode(int sessionAcknowledgeMode)
    设置确认模式,sessionAcknowledgeMode可以取javax.jms.Session.AUTO_ACKNOWLEDGE(默认),javax.jms.Session.CLIENT_ACKNOWLEDGE,javax.jms.Session.DUPS_OK_ACKNOWLEDGE。
  • public void setSessionAcknowledgeModeName(String constantName)
    通过名字来设置确认模式,默认为"AUTO_ACKNOWLEDGE"。

4.事务

下面的API,用来设置事务:
  • public void setSessionTransacted(boolean sessionTransacted)
    设置是否使用事务,默认为false。
 
更多DefaultMessageListenerContainer的相关配置,参考其API。

精简的配置

称之为精简版的配置,因为一个容器可以配置多个监听器。
 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> 
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:
 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 } 

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在可信任包内,才能被提取出来。
你可以这样配置可信任包:
 1 <bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory"> 
 2     <property name="brokerURL" value="tcp://localhost:61616" /> 
 3     <property name="trustedPackages"> 
 4         <list> 
 5             <!-- you must add the package or parent-package of object which is put in ObjectMessage --> 
 6             <value>java.lang</value> 
 7             <value>java.sql</value> 
 8             <value>cn.sinobest</value> 
 9         </list> 
10     </property> 
11 </bean>
关于ObjectMessage安全性的说明,参考 IT虾米网

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进行包装,建议用于测试或者单机的环境。
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>

3.PooledConnectionFactory

使用org.apache.activemq.pool.PooledConnectionFactory对ActiveMQConnectionFactory进行包装,暂不知有什么优化。
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>

分布式事务


首先引入数据源的概念,数据源能提供数据或者能存储数据,一般是两个特点都具备。如数据库、消息队列,都是数据源。
如果你要使用多个数据源,那就要考虑分布式事务。比如从broker-A中接收消息,发送到broker-B。如果在发送的过程中发生错误,接收的消息就不应该确认,否则会被 broker-A移除,造成消息的丢失。或者你想从 broker-A中接收消息,写入数据库,也有同样的问题。而分布式事务要处理的问题,就是涉及到多个数据源的事务问题,保证涉及多个数据源的操作要么同时成功,要么同时失败。
关于分布式事务的讨论,可选的参考 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框架。
事务的配置:
 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     <!-- part-1 --> 
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     <!-- part-2 --> 
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>

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

 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 }

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

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

2.broker-B到数据库

从broker-B中接收消息,写入数据库。这一次没有使用Atomikos,也实现了当写库失败时回滚消息的事务效果;而前一个小节broker-A到broker-B,如果不使用Atomikos就无法达到事务效果。或许因为它使用的是同步接收,而接下来我们使用的是异步接收。接收的方式是否有对分布式事务有关系,结论还不能确定。
在这里,我们甚至没有再使用spring的JtaTransactionManager。
 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:jms="http://www.springframework.org/schema/jms" 
 5     xsi:schemaLocation="http://www.springframework.org/schema/beans 
 6         http://www.springframework.org/schema/beans/spring-beans-4.2.xsd 
 7         http://www.springframework.org/schema/jms 
 8         http://www.springframework.org/schema/jms/spring-jms-4.2.xsd"> 
 9     <bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory"> 
10         <property name="brokerURL" value="${java.naming.provider.url}" /> 
11     </bean> 
12      
13     <bean id="converteredLogListener" class="cn.sinobest.asj.log.listener.ConverteredLogListener" /> 
14      
15     <jms:listener-container connection-factory="connectionFactory" 
16         concurrency="1" acknowledge="transacted"> 
17         <jms:listener destination="asj.log" ref="converteredLogListener" 
18             method="onLog" /> 
19     </jms:listener-container> 
20 </beans>

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






发布评论
IT源码网

微信公众号号:IT虾米 (左侧二维码扫一扫)欢迎添加!

AMQ学习笔记 - 12. Spring-JmsTemplate特性设置讲解
你是第一个吃螃蟹的人
发表评论

◎欢迎参与讨论,请在这里发表您的看法、交流您的观点。