利用webservice和jms实现系统间的数据同步之二

接上一篇文章

此系统采用ActiveMQ作为jms消息服务器
jmsserver系统的结构图如下：

?

jdbc.properties配置如下：

jdbc.driverClassName=oracle.jdbc.xa.client.OracleXADataSource
jdbc.url=jdbc:oracle:thin:@10.78.124.76:1521/test
jdbc.username=oracle
jdbc.password=oracle
jdbc.poolSize=1
jdbc.maxPoolSize=30
hibernate.dialect=org.hibernate.dialect.Oracle9Dialect

?

spring配置文件如下：

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
	xmlns:context="http://www.springframework.org/schema/context"
	xmlns:jaxws="http://cxf.apache.org/jaxws"
	xmlns:jaxrs="http://cxf.apache.org/jaxrs"
	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xmlns:aop="http://www.springframework.org/schema/aop"
	xmlns:tx="http://www.springframework.org/schema/tx"
	xsi:schemaLocation="http://www.springframework.org/schema/beans 
	http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
	http://www.springframework.org/schema/context
	http://www.springframework.org/schema/context/spring-context-3.0.xsd
	http://cxf.apache.org/jaxws 
	http://cxf.apache.org/schemas/jaxws.xsd
	http://cxf.apache.org/jaxrs 
	http://cxf.apache.org/schemas/jaxrs.xsd
	http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-3.0.xsd
    http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-3.0.xsd">

	<import resource="classpath:META-INF/cxf/cxf.xml" />
	<import resource="classpath:META-INF/cxf/cxf-extension-soap.xml" />
	<import resource="classpath:META-INF/cxf/cxf-servlet.xml" />

	<context:component-scan base-package="com.test" />

	<!-- webserice接收客户端 -->
	<jaxws:client id="userService"
		address="http://10.78.69.50:8088/webserviceserver/service/user"
		serviceClass="com.test.UserService" />

	<!-- restful接收客户端，scope的状态需设置为原型模式 -->
	<bean id="restfulClient"
		class="org.apache.cxf.jaxrs.client.WebClient" factory-method="create"
		scope="prototype">
		<constructor-arg type="java.lang.String"
			value="http://10.78.69.50:8088/webserviceserver/service/restful" />
	</bean>

	<!-- 加载jdbc文件 -->
	<bean id="propertyConfig"
		class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
		<property name="locations">
			<list>
				<value>classpath:jdbc.properties</value>
			</list>
		</property>
	</bean>

	<bean id="jndiTemplate"
		class="org.springframework.jndi.JndiTemplate">
		<property name="environment">
			<props>
				<prop key="java.naming.factory.initial">
					org.apache.activemq.jndi.ActiveMQInitialContextFactory
				</prop>
				<prop key="java.naming.provider.url">tcp://10.78.69.50:61616</prop>
				<prop key="java.naming.security.principal">system</prop>
				<prop key="java.naming.security.credentials">manager</prop>
			</props>
		</property>
	</bean>

	<!-- jndi连接工厂 -->
	<bean id="jndiConnectionFactory"
		class="org.springframework.jndi.JndiObjectFactoryBean">
		<property name="jndiTemplate" ref="jndiTemplate" />
		<property name="jndiName" value="QueueCF" />
	</bean>

	<!-- jms session缓存 -->
	<bean id="queueConnectionFactory"
		class="org.springframework.jms.connection.CachingConnectionFactory">
		<property name="targetConnectionFactory"
			ref="jndiConnectionFactory" />
		<property name="sessionCacheSize" value="10" />
	</bean>

	<bean id="destinationResolver"
		class="org.springframework.jms.support.destination.JndiDestinationResolver">
		<property name="jndiTemplate" ref="jndiTemplate" />
		<property name="cache" value="true" />
		<property name="fallbackToDynamicDestination" value="false" />
	</bean>

    <!-- 用atomikos实现数据源，使用jta事务管理 -->
    <bean id="xaDataSource" class="com.atomikos.jdbc.AtomikosDataSourceBean"
		init-method="init" destroy-method="close">
		<property name="uniqueResourceName" value="oracle_ds" />
		<property name="xaDataSourceClassName" value="${jdbc.driverClassName}" />
		<property name="xaProperties">
			<props>    
                <prop key="URL">${jdbc.url}</prop>
               	<prop key="user">${jdbc.username}</prop>    
                <prop key="password">${jdbc.password}</prop> 
           </props>
		</property>
		<property name="poolSize" value="${jdbc.poolSize}"/>
        <property name="maxPoolSize" value="${jdbc.maxPoolSize}"/>
	</bean>
	
    <bean id="sessionFactory"
		class="org.springframework.orm.hibernate3.LocalSessionFactoryBean">
		<property name="dataSource">
			<ref local="xaDataSource" />
		</property>
		<property name="mappingLocations">
			<value>
				classpath:com/test/model/*.hbm.xml
			</value>
		</property>
		<property name="hibernateProperties">
			<props>
				<prop key="hibernate.dialect">${hibernate.dialect}</prop>
				<prop key="hibernate.show_sql">true</prop>
				<prop key="hibernate.jdbc.fetch_size">50</prop>
				<prop key="hibernate.jdbc.batch_size">25</prop>
				<prop key="hibernate.cache.use_second_level_cache">false</prop>
			</props>
		</property>
	</bean>
    
    <bean id="atomikosUserTransaction"
		class="com.atomikos.icatch.jta.UserTransactionImp">
		<property name="transactionTimeout" value="300"/>
	</bean>
	<bean id="atomikosTransactionManager"
		class="com.atomikos.icatch.jta.UserTransactionManager"
		init-method="init" destroy-method="close">
		<property name="forceShutdown">
			<value>true</value>
		</property>
	</bean>
	
	<bean id="transactionManager"
		class="org.springframework.transaction.jta.JtaTransactionManager">
		<property name="transactionManager">
			<ref bean="atomikosTransactionManager" />
		</property>
		<property name="userTransaction">
			<ref bean="atomikosUserTransaction" />
		</property>
	</bean>
	
	<!-- 设置事务的传播行为--> 
	<tx:advice id="txAdvice" transaction-manager="transactionManager" >
		<tx:attributes>
			<tx:method name="find*" propagation="SUPPORTS" read-only="true"/>
			<tx:method name="get*"  propagation="SUPPORTS" read-only="true"/>
			<tx:method name="has*" 	propagation="SUPPORTS" read-only="true"/>
			<tx:method name="is*" 	propagation="SUPPORTS" read-only="true"/>
			
			<tx:method name="*" 		propagation = "REQUIRED" rollback-for="com.test.service.MyException"/>
			<tx:method name="insert*" 	propagation = "REQUIRED" rollback-for="com.test.service.MyException"/>
			<tx:method name="update*" 	propagation = "REQUIRED" rollback-for="com.test.service.MyException"/>
			<tx:method name="modify*" 	propagation = "REQUIRED" rollback-for="com.test.service.MyException"/>
			<tx:method name="delete*" 	propagation = "REQUIRED" rollback-for="com.test.service.MyException"/>
			<tx:method name="save*" 	propagation = "REQUIRED" rollback-for="com.test.service.MyException"/>
		</tx:attributes>
	</tx:advice>
	
	<!-- 事务通知 -->
	<aop:config>
		<aop:advisor advice-ref="txAdvice" pointcut="execution(* com.test.service..*.*(..))" />
	</aop:config>

    <!-- 使用spring的jms发送消息 -->
	<bean id="jmsTemplate"
		class="org.springframework.jms.core.JmsTemplate">
		<property name="connectionFactory" ref="queueConnectionFactory" />
		<property name="defaultDestinationName" value="queue1" />
		<property name="destinationResolver" ref="destinationResolver" />
		<property name="pubSubDomain" value="false" />
		<!-- 等待消息的时间(ms) -->
		<property name="receiveTimeout" value="3000" />
		<!-- 开启事务，注意此处的值 -->
		<property name="sessionTransacted" value="true" />
	</bean>
	
	<bean id="syncDao" class="com.test.dao.SyncDaoImpl">
	    <property name="sessionFactory" ref="sessionFactory" />
	    <property name="jdbcTemplate" ref="jdbcTemplate" />
	</bean>
	
	<bean id="jdbcTemplate" class="org.springframework.jdbc.core.JdbcTemplate">
	    <property name="dataSource" ref="xaDataSource"/>
	</bean>
</beans>

web.xml文件如下：

?

<?xml version="1.0" encoding="UTF-8"?>
<web-app version="2.5" xmlns="http://java.sun.com/xml/ns/javaee"
	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="http://java.sun.com/xml/ns/javaee 
	http://java.sun.com/xml/ns/javaee/web-app_2_5.xsd">

	<context-param>
		<param-name>contextConfigLocation</param-name>
		<param-value>classpath*:application-*.xml</param-value>
	</context-param>
	<listener>
		<listener-class>
			org.springframework.web.context.ContextLoaderListener
		</listener-class>
	</listener>

    <filter>
	    <filter-name>struts2</filter-name>
	    <filter-class>org.apache.struts2.dispatcher.ng.filter.StrutsPrepareAndExecuteFilter</filter-class>
	</filter>
	
	<filter-mapping>
	    <filter-name>struts2</filter-name>
	    <url-pattern>/*</url-pattern>
	</filter-mapping>

	<welcome-file-list>
		<welcome-file>index.jsp</welcome-file>
	</welcome-file-list>
</web-app>

SyncDataAction内如如下：

?

package com.test;

import org.apache.cxf.jaxrs.client.WebClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Controller;

import com.opensymphony.xwork2.ActionSupport;
import com.test.service.SyncDataService;

@Controller("syncDataAction")
@Scope("prototype")
public class SyncDataAction extends ActionSupport
{

    private static final long serialVersionUID = -7231587168221074432L;
    
    @Autowired
    private SyncDataService syncDataService;
    
    @Autowired
    private UserService userService;
    
    @Autowired
    private WebClient restfulClient;
    
    /**
     * 这个action的功能是从webserviceserver系统中通过webservice获取数据
     * 并把这些数据通过jms发送出去
     * 同时保存数据到数据库
     * 具体详见SyncDataServiceImpl的syncData方法
     */
    public String sync()
    {
        try
        {
            syncDataService.syncData("queue1",userService.getUserStr().toString());
        }
        catch(Exception e)
        {
            e.printStackTrace();
        }
        return SUCCESS;
    }
    
    /**
     * 这个action的功能是从webserviceserver系统中通过restful webservice获取数据
     */
    public String restfulSync()
    {
        try
        {
            syncDataService.restfulSyncData("queue2",restfulClient.path("/caoguowei").get(String.class));
        }
        catch(Exception e)
        {
            e.printStackTrace();
        }
        
        return SUCCESS;
    }

}

UserService内如如下：

?

package com.test;

import java.util.List;

import javax.jws.WebResult;
import javax.jws.WebService;

/**
 * 从webserviceservice系统调用的webservice接口
 */
@WebService
public interface UserService
{
    @WebResult List<User> getUserList();
    
    User getUserStr();
}

SyncDataServiceImpl内容如下：

?

package com.test.service;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.stereotype.Service;

import com.test.dao.SyncDao;

@Service
public class SyncDataServiceImpl implements SyncDataService
{

    @Autowired
    private JmsTemplate jmsTemplate;
    
    @Autowired
    private SyncDao syncDao;
    
    /**
     * 如果发送jms消息失败，或保存数据到数据库失败，则事务回滚，两者都失败
     */
    public void syncData(String queue,Object message) throws Exception
    {
        //发送jms消息
        jmsTemplate.convertAndSend(queue,message);
        
        //保存数据到数据库
        syncDao.insertUser(message.toString());
    }

    public void restfulSyncData(String queue,Object message) throws Exception
    {
        jmsTemplate.convertAndSend(queue,message);
    }
}

jndi.properties内容如下：

?

java.naming.factory.initial = org.apache.activemq.jndi.ActiveMQInitialContextFactory
java.naming.provider.url = tcp://10.78.69.50:61616
java.naming.security.principal = system
java.naming.security.credentials = manager
connectionFactoryNames = QueueCF
queue.queue1=jms.queue1
queue.queue2=jms.queue2

jta.properties内容如下：

com.atomikos.icatch.service=com.atomikos.icatch.standalone.UserTransactionServiceFactory
com.atomikos.icatch.log_base_name = tmlog
com.atomikos.icatch.tm_unique_name = com.atomikos.spring.jms.tm
com.atomikos.icatch.console_file_name = tm.out 
com.atomikos.icatch.console_log_level=INFO

另外还需要在ActiveMQ的activemq.xml文件中定义两个队列：

<destinations>    
    <queue name="queue1" physicalName="jms.queue1" />    
    <queue name="queue2" physicalName="jms.queue2" />
</destinations>

activemq.xml文件的位置在ActiveMQ安装目录的conf文件夹下，开启服务前要首先气功ActiveMQ

服务，可以通过运行bin目录下的activemq.bat命令启动服务。