ActiveMQ与spring集成实例之使用消息转换器

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="http://www.springframework.org/schema/beans
    
    http://www.springframework.org/schema/beans/spring-beans-2.0.xsd">
	<!--创建连接工厂 -->
	<bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
		<property name="brokerURL" value="tcp://localhost:61616"></property>
	</bean>
	<!-- 声明ActiveMQ消息目标,目标可以是一个队列，也可以是一个主题ActiveMQTopic -->
	<bean id="destination" class="org.apache.activemq.command.ActiveMQQueue">
		<constructor-arg index="0"
			value="com.acca.activemq"></constructor-arg>
	</bean>
    <!---->
	<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
		<property name="connectionFactory" ref="connectionFactory"></property>
		<property name="defaultDestination" ref="destination"></property>
		<property name="receiveTimeout" value="600"></property>
        <property name="messageConverter" ref="myConvert"></property>
	</bean>
	<bean id="sender" class="com.acca.activemq.Sender">
		<property name="jmsTemplate" ref="jmsTemplate"></property>

	</bean>
	<bean id="receiver" class="com.acca.activemq.Receiver">
		<property name="jmsTemplate" ref="jmsTemplate"></property>
	</bean>
	<!-- 消息转换器 -->
	<bean id="myConvert" class="com.acca.activemq.convert.MyConvert">
	</bean>
	

</beans>

 

/*
 * Copyright (c) 2012-2032 Accounting Center of China Aviation(ACCA).
 * All Rights Reserved.
 */
package com.acca.activemq.convert;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.ObjectMessage;
import javax.jms.Session;

import org.apache.activemq.command.ActiveMQObjectMessage;
import org.springframework.jms.support.converter.MessageConversionException;
import org.springframework.jms.support.converter.MessageConverter;

import com.acca.activemq.entity.User;

/**
 * 
 * 消息转换器
 * 
 * @author zhouhua, 2012-12-25
 */
public class MyConvert implements MessageConverter {

    /**
     * @param arg0
     * @return
     * @throws JMSException
     * @throws MessageConversionException
     * @see org.springframework.jms.support.converter.MessageConverter#fromMessage(javax.jms.Message)
     */
    @Override
    public Object fromMessage(Message message) throws JMSException, MessageConversionException {
        User user=null;
            if(message instanceof ActiveMQObjectMessage){
                ActiveMQObjectMessage aMsg = (ActiveMQObjectMessage) message; 
                 user=(User) aMsg.getObject();
            }
        return user;
    }

    /**
     * @param arg0
     * @param arg1
     * @return
     * @throws JMSException
     * @throws MessageConversionException
     * @see org.springframework.jms.support.converter.MessageConverter#toMessage(java.lang.Object, javax.jms.Session)
     */
    @Override
    public Message toMessage(Object object, Session session) throws JMSException,
        MessageConversionException {
        
        ActiveMQObjectMessage msg = (ActiveMQObjectMessage) session.createObjectMessage();
        msg.setObject((User) object);
        return msg; 
    }

}

 

/*
 * Copyright (c) 2012-2032 Accounting Center of China Aviation(ACCA).
 * All Rights Reserved.
 */
package com.acca.activemq;

import javax.jms.JMSException;
import javax.jms.MapMessage;

import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.support.JmsUtils;

import com.acca.activemq.entity.User;

/**
 * 
 * 
 * 
 * @author zhouhua, 2012-12-25
 */
public class Receiver {
    private JmsTemplate jmsTemplate;

    // getter and setter
    public JmsTemplate getJmsTemplate() {
        return jmsTemplate;
    }

    public void setJmsTemplate(JmsTemplate jmsTemplate) {
        this.jmsTemplate = jmsTemplate;
    }

    /**
     * 构造函数
     */
    public Receiver() {
    }

    public User receiveMessage() {
        User user=null;
        user=(User) jmsTemplate.receiveAndConvert();
        return user;
    }
}

 

/*
 * Copyright (c) 2012-2032 Accounting Center of China Aviation(ACCA).
 * All Rights Reserved.
 */
package com.acca.activemq;

import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

import com.acca.activemq.entity.User;

/**
 * 
 * 
 * 
 * @author zhouhua, 2012-12-25
 */
public class ReceiverTest {

    /**
     * @param args
     */
    public static void main(String[] args) {
        // TODO Auto-generated method stub
        ApplicationContext context = new ClassPathXmlApplicationContext("applicationContext.xml");
        Receiver receiver = (Receiver) context.getBean("receiver");
        User user=receiver.receiveMessage();
        System.out.print("地址："+user.getAddress()+"姓名:"+user.getUserName());
    }

}

 

/*
    /*
 * Copyright (c) 2012-2032 Accounting Center of China Aviation(ACCA).
 * All Rights Reserved.
 */
package com.acca.activemq;

import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.Session;

import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;

import com.acca.activemq.entity.User;

/**
 * 
 * 
 * 
 * @author zhouhua, 2012-12-25
 */
public class Sender {

    private JmsTemplate jmsTemplate;

    // getter and setter
    public JmsTemplate getJmsTemplate() {
        return jmsTemplate;
    }

    public void setJmsTemplate(JmsTemplate jmsTemplate) {
        this.jmsTemplate = jmsTemplate;
    }

    public void sendInfo() {
        User user=new User();
        user.setAddress("北京市");
        user.setUserName("张三");
        jmsTemplate.convertAndSend(user);
    }
}

 

/*
 * Copyright (c) 2012-2032 Accounting Center of China Aviation(ACCA).
 * All Rights Reserved.
 */
package com.acca.activemq;

import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

/**
 * 
 * 
 *
 * @author zhouhua, 2012-12-25
 */
public class SenderTest {

    /**
     * @param args
     */
    public static void main(String[] args) {
        ApplicationContext context = new ClassPathXmlApplicationContext("applicationContext.xml");
        Sender sender = (Sender) context.getBean("sender");
        sender.sendInfo();
    }

}

 

/*
 * Copyright (c) 2012-2032 Accounting Center of China Aviation(ACCA).
 * All Rights Reserved.
 */
package com.acca.activemq.entity;

import java.io.Serializable;

/**
 * 
 * 
 *
 * @author zhouhua, 2012-12-25
 */
public class User implements Serializable {

    private String userName;
    private String address;

    /**
     * @return the userName
     */
    public String getUserName() {
        return userName;
    }

    /**
     * @param userName the userName to set
     */
    public void setUserName(String userName) {
        this.userName = userName;
    }

    /**
     * @return the address
     */
    public String getAddress() {
        return address;
    }

    /**
     * @param address the address to set
     */
    public void setAddress(String address) {
        this.address = address;
    }

}

 

消息转换器主要目的将消息以pojo的形式传送，消息转换器在配置文件中配置好即可，JmsTemplete会自动根据类型转换