ActiveMQ 结合Spring进行数据同步

注意事项hibernate配置文件必须设置自动提交否则不能插入成功

配置文件

<?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:aop="http://www.springframework.org/schema/aop"
		xmlns:tx="http://www.springframework.org/schema/tx"
		xmlns:context="http://www.springframework.org/schema/context"
		xsi:schemaLocation="
			http://www.springframework.org/schema/beans 
			http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
			http://www.springframework.org/schema/aop 
			http://www.springframework.org/schema/aop/spring-aop-2.5.xsd
			http://www.springframework.org/schema/tx 
			http://www.springframework.org/schema/tx/spring-tx-2.5.xsd
			http://www.springframework.org/schema/context 
			http://www.springframework.org/schema/context/spring-context-2.5.xsd"> 
		
    <!-- 配置activeMQ -->
    <bean id="jmsFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
        <property name="brokerURL" value="tcp://localhost:61616" />
        <property name="userName" value="system" />  
        <property name="password" value="manager" />  
    </bean>

    <!-- queue目的地配置 -->  
    <bean id="destination" class="org.apache.activemq.command.ActiveMQQueue">  
        <constructor-arg index="0" value="spring-queue" />  
    </bean>  
    <!-- topic目的地配置，其实不管是topic还是queue则他们的底层实现不同但是通过封装api就差不多了，而在spring中更是简单 -->  
    <bean id="destinationTopic" class="org.apache.activemq.command.ActiveMQTopic">  
        <constructor-arg index="0" value="spring-topic" />  
    </bean>  
  
    <!-- spring 使用jmsTemplate来实现消息的发送和接受 -->  
    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">  
        <property name="connectionFactory" ref="jmsFactory"></property>  
        <property name="defaultDestination" ref="destinationTopic"></property>
	<!--pubSubDomain设置true则是Topic模式 false则是queue模式 -->
        <property name="pubSubDomain" value="true"/>
    </bean>
    
    <!--异步监听 -->  
    <bean id="myMessageListener" class="xxx.activemq.MyMessageListener">
    	  <property name="commonDao">
			<ref bean="commonDao" />
		</property>
    </bean> 
    <!-- 主题监听容器 {Topic模式} -->
    <bean id="jmsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer" lazy-init="false">
        <property name="connectionFactory" ref="jmsFactory" />  
        <property name="destination" ref="destinationTopic" />  
        <property name="messageListener" ref="myMessageListener" />
        <property name="receiveTimeout" value="1000000" />
    </bean>
    
</beans>

监听类

public class MyMessageListener implements MessageListener {
	
	private static final Log LOG = LogFactory.getLog(MyMessageListener.class);
	
	private CommonDao commonDao;
	
	public CommonDao getCommonDao() {
		return commonDao;
	}

	public void setCommonDao(CommonDao commonDao) {
		this.commonDao = commonDao;
	}

	@SuppressWarnings("all")
    public void onMessage(Message arg0) {
        try {
        	ObjectMessage message =(ObjectMessage) arg0;
        	System.out.println(message);
        	if(message != null){
            	HashMap msg = (HashMap)message.getObject();
            	this.handleMessage(msg.get("MSGBODY"), (String)msg.get("HANDLETYPE"));
            	LOG.info("ActiveMQ成功处理一条数据！类名---->"+msg.get("CLASSNAME"));
            }
        } catch (Exception e) {  
            e.printStackTrace();  
        }  
    }  
	
	private void handleMessage(Object obj,String type){
		if(obj == null || type == null){
			LOG.error("ActiveMQ处理的消息体不能为空或处理类型不能为空！");
			throw new NullPointerException("ActiveMQ处理的消息体不能为空！");
		}
		if(HANDLETYPE.ADD.toString().equals(type)){
			this.commonDao.save(obj);
		}
		if(HANDLETYPE.UPDATE.toString().equals(type)){
			this.commonDao.update(obj);
		}
		if(HANDLETYPE.DELETE.toString().equals(type)){
			this.commonDao.delete(obj);
		}
	}
}


enum HANDLETYPE{
	ADD,UPDATE,DELETE
}

消息发送类

@Service("activemqMessageService")
@SuppressWarnings("all")
public class ActivemqMessageServiceImpl implements ActivemqMessageService{

	private static final Log log = LogFactory.getLog(DataProviderServiceImpl.class);
	@Resource
    private JmsTemplate jmsTemplate;
	
	@Override
	public void sendMessage(Object msgObj,String handleType,String className){
		String destination =  jmsTemplate.getDefaultDestination().toString();
		final HashMap<String, Object> map = new HashMap<String, Object>();
        map.put("HANDLETYPE", handleType);  
        map.put("MSGBODY", msgObj);
        map.put("CLASSNAME", className);
        System.out.println("向队列" +destination+ "发送了消息------------" + map);
        jmsTemplate.send(new MessageCreator() {
            public Message createMessage(Session session) throws JMSException {
                return session.createObjectMessage(map);
            }
        });
	}
	
}