Apusic+SpringJMS持久订阅配置及Topic重启后不能接受到信息问题解决

Spring JMS是一个强大的组件，允许用户通过配置的方式进行JMS的订阅/接受，下面开始动手建立起来。

1.Spring config 配置文件

<?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:jms="http://www.springframework.org/schema/jms"

xsi:schemaLocation="
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd http://www.springframework.org/schema/jms http://www.springframework.org/schema/jms/spring-jms-2.5.xsd"
default-lazy-init="true" default-autowire="byName" >

<!--JNDI 模板配置，用于定于基本的连接工厂，IP，用户/密码等环境属性

java.naming.factory.initial=com.apusic.naming.jndi.CNContextFactory

java.naming.factory.url.pkgs=com.apusic.naming.jndi.url-->

<bean id="jndiTemplate" class="org.springframework.jndi.JndiTemplate">

<property name="environment">

<props>

<prop key="java.naming.factory.initial">

com.apusic.naming.jndi.CNContextFactory

</prop>

<prop key="java.naming.provider.url">

iiop://192.168.11.62:7777

</prop>

<prop key="java.naming.factory.url.pkgs">

com.apusic.naming.jndi.url

</prop>

<prop key="UserName">

test

</prop>

<prop key="Password" >

test

</prop>

</props>

</property>

</bean>

<!-- JNDI工厂类定义，用于获取JNDI对象，指定JMS类型是Topic还是Quence

jms ConnectionFactory -->

<bean id="jmsFactory"

class="org.springframework.jndi.JndiObjectFactoryBean">

<property name="jndiTemplate">

<ref bean="jndiTemplate"/>

</property>

<property name="jndiName">

<value>jms/ConnectionFactory</value>

</property>

<property name="expectedType" value="javax.jms.TopicConnectionFactory" />

</bean>

<!-- SpringJMS连接工厂创建类，用于实例化JNDI连接对象，clientId是必须输入的属性，reconnectOnException是失败重新连接的属性Spring JMS推荐的类为SingleConnectionFactory，此处创建子类MySingleConnectionFactory是为了解决Topic服务器被重启后，JMS是不能正常接受到信息的，重新启动也不想，主要是clientId没有被改变导致重连接了但订阅存在问题，找了些参数配置没有找到合适的属性，于是通过子类来进行clientId的重新设定，clientId重新设定后Topic重启也是可以接受到信息的。
-->

<bean id="singleConnectionFactory"

class="from.sm.to.wm.jms.factory.MySingleConnectionFactory" >

<property name="targetConnectionFactory" ref="jmsFactory" />

<property name="reconnectOnException" value="true" />

<property name="clientId" >

<value>Asn-Receiver</value>

</property>

</bean>

<!-- 此处定义JMS模板 -->

<bean id="jmsTemplate"

class="org.springframework.jms.core.JmsTemplate">

<property name="connectionFactory">

<ref bean="singleConnectionFactory"/>

</property>

<property name="defaultDestination">

<ref bean="sendDestination"/>

</property>

<property name="receiveTimeout">

<value>50000</value>

</property>

<property name="pubSubDomain" value="true" />

<property name="explicitQosEnabled" value="true" />

</bean>

<!-- 此处定义JMS信息发送端，jndiName实际上是指Topic-->

<bean id="sendDestination"

class="org.springframework.jndi.JndiObjectFactoryBean">

<property name="jndiTemplate">

<ref bean="jndiTemplate"/>

</property>

<property name="expectedType" value="javax.jms.Topic"/>

<!-- 定义jms发收消息，这里我用主题模式来发消息 -->

<property name="jndiName" >

<value>FrontToBack</value>

</property>

<property name="lookupOnStartup" value="true"/>

</bean>

<!-- 此处定义JMS信息接受端，jndiName实际上是指Topic-->

<bean id="receiveDestination"

class="org.springframework.jndi.JndiObjectFactoryBean">

<property name="jndiTemplate">

<ref bean="jndiTemplate"/>

</property>

<property name="jndiName">

<value>Test_Private_TO3PL</value>

</property>

<property name="expectedType" value="javax.jms.Topic"/>

<property name="lookupOnStartup" value="true"/>

</bean>

<!-- 此处定义JMS信息实际发送对象-->

<bean id="jmsSender" class="from.sm.to.wm.jms.JMSSender" >

<property name="jmsTemplate">

<ref bean="jmsTemplate"/>

</property>

</bean>

<!-- 此处定义JMS信息组装并实行发送-->

<bean id="singalMessageService" class="from.sm.to.wm.jms.SingalMessageService">

<property name="sender" ref="jmsSender" />

</bean>

<!-- 此处定义JMS信息接受Listener-->

<bean id="jmsListener" class="from.sm.to.wm.jms.listener.AsnMessageListener">

<property name="singalMessageService" ref="singalMessageService"/>

</bean>

<!-- 此处定义JMS信息接受Listener的错误绑定-->

<bean id="jmsErrorHandler" class="from.sm.to.wm.jms.handler.JMSDefaultErrorHandler"/>

<!-- 此处定义JMS信息接受Listener的异常Listener-->

<bean id="jmsExceptionListener" class="from.sm.to.wm.jms.listener.JMSDefaultExceptionListener"/>

<!-- 指明了jmsTopic队列的接收监听器 -->

<bean id="listenercontainer"

class="org.springframework.jms.listener.DefaultMessageListenerContainer">

<property name="connectionFactory" ref="singleConnectionFactory"></property>

<property name="destination" ref="receiveDestination"></property>

<property name="messageListener" ref="jmsListener"></property>

<property name="subscriptionDurable" value="true"></property>

<property name="cacheLevel" value="4"></property>

<property name="pubSubDomain" value="true"/>

<property name="concurrentConsumers" value="1"/>

<!-- 此处订阅JMS是自动启动的-->

<property name="autoStartup" value="true"/>

<property name="sessionAcknowledgeMode" value="1" />

<property name="errorHandler" ref="jmsErrorHandler" />

<property name="exceptionListener" ref="jmsExceptionListener" />

<!-- 此处定义消息选择器 -->

<property name="messageSelector" value="MessageTypeCode='1B1' AND ToRole='Back' AND FromRole='BPSFront'"/>

</bean>

</beans>

2.MySingleConnectionFactory对象的核心处理，复写getSharedConnectionProxy方法

@Override

protected Connection getSharedConnectionProxy(Connection target) {

Connection conn = super.getSharedConnectionProxy(target);

if (conn != null) {

// 此处解决持久订阅由于Topic服务器被关闭即使重新连接了也不能正常接受到Topic内容

// 判断可能是由于即使重新连接，clientId也是相同的，导致每次的重新连接相对于应用内部还是旧的连接Session

// 因此在此处进行了重新设定clientId，测试结果OK，此处为关键所在

String clientid = getClientId() + new Random().nextInt(5000);

super.setClientId(clientid);

Logger.getLogger(MySingleConnectionFactory.class).debug("reconnection by difference client id " + clientid);

try {

Method setClientID = conn.getClass().getMethod("setClientID", String.class);

setClientID.setAccessible(true);

setClientID.invoke(conn, clientid);

} catch (NoSuchMethodException ex) {

Logger.getLogger(MySingleConnectionFactory.class).debug(ex);

} catch (SecurityException ex) {

Logger.getLogger(MySingleConnectionFactory.class).debug(ex);

} catch (IllegalAccessException ex) {

Logger.getLogger(MySingleConnectionFactory.class).debug(ex);

} catch (IllegalArgumentException ex) {

Logger.getLogger(MySingleConnectionFactory.class).debug(ex);

} catch (InvocationTargetException ex) {

Logger.getLogger(MySingleConnectionFactory.class).debug(ex);

}

}

return conn;

}