ActiveMQ之spring集成消息转换器MessageConverter

àì夳堔傛蜴生んèń 2022-08-05 02:26 230阅读 0赞

MessageConverter的作用主要有两方面，一方面它可以把我们的非标准化Message对象转换成我们的目标Message对象，这主要是用在发送消息的时候；另一方面它又可以把我们的Message对象转换成对应的目标对象，这主要是用在接收消息的时候。

下面我们就拿发送一个对象消息来举例， 传输USer对象：

package cn.slimsmart.activemq.demo.spring.msgconverter;
    
    import java.io.Serializable;
    import java.util.Date;
    
    public class User implements Serializable{
    	
    	private static final long serialVersionUID = 1L;
    	
    	private String id;
    	private String name;
    	private int age;
    	private Date birthDate;
    	
    	public String getId() {
    		return id;
    	}
    	public void setId(String id) {
    		this.id = id;
    	}
    	public String getName() {
    		return name;
    	}
    	public void setName(String name) {
    		this.name = name;
    	}
    	public int getAge() {
    		return age;
    	}
    	public void setAge(int age) {
    		this.age = age;
    	}
    	public Date getBirthDate() {
    		return birthDate;
    	}
    	public void setBirthDate(Date birthDate) {
    		this.birthDate = birthDate;
    	}
    }

定义转换器

package cn.slimsmart.activemq.demo.spring.msgconverter;
    
    import java.io.Serializable;
    
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.ObjectMessage;
    import javax.jms.Session;
    
    import org.springframework.jms.support.converter.MessageConversionException;
    import org.springframework.jms.support.converter.MessageConverter;
    
    /**
     * spring 消息转换器
     * @author slimina
     *
     */
    public class UserMessageConverter implements MessageConverter {
    
    	public Message toMessage(Object object, Session session)  
                throws JMSException, MessageConversionException {  
            return session.createObjectMessage((Serializable) object);  
        }  
       
        public Object fromMessage(Message message) throws JMSException,  
                MessageConversionException {  
            ObjectMessage objMessage = (ObjectMessage) message;  
            return objMessage.getObject();  
        }
    }

消息接收

package cn.slimsmart.activemq.demo.spring.msgconverter;
    
    import com.google.gson.Gson;
    
    public class UserMessageListener {
    
    	public void receiveMessage(User user) {
    		System.out.println(new Gson().toJson(user));
    	}
    	
    	 public void receiveMessage(String message) {  
    	        System.out.println("接收到一个纯文本消息，消息内容是：" + message);  
    	 }  
    }

生产者配置：

<?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:p="http://www.springframework.org/schema/p"
    	xmlns:aop="http://www.springframework.org/schema/aop" xmlns:context="http://www.springframework.org/schema/context"
    	xmlns:jee="http://www.springframework.org/schema/jee" xmlns:jms="http://www.springframework.org/schema/jms"
    	xmlns:tx="http://www.springframework.org/schema/tx" xmlns:util="http://www.springframework.org/schema/util"
    	xmlns:lang="http://www.springframework.org/schema/lang"
    	xsi:schemaLocation="
    			http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-3.0.xsd
    			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://www.springframework.org/schema/jee http://www.springframework.org/schema/jee/spring-jee-3.0.xsd
    			http://www.springframework.org/schema/jms http://www.springframework.org/schema/jms/spring-jms-3.0.xsd
    			http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-3.0.xsd
    			http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util-3.0.xsd
    			http://www.springframework.org/schema/lang http://www.springframework.org/schema/lang/spring-lang-3.0.xsd">
    
    	   
        <bean id="connectionFactory"  
            class="org.springframework.jms.connection.CachingConnectionFactory">  
              
            <property name="sessionCacheSize" value="10" />  
            <property name="targetConnectionFactory">  
                <bean class="org.apache.activemq.ActiveMQConnectionFactory">  
                      
                    <property name="brokerURL" value="tcp://192.168.18.43:61616" />  
                       
                    <property name="useAsyncSend" value="false" />  
                </bean>  
            </property>  
        </bean> 
    
     	
    	<bean id="testQueue" class="org.apache.activemq.command.ActiveMQQueue">
    		<property name="physicalName" value="spring.queue.converte" />
    	</bean>
        
    	
    	<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
    		<property name="connectionFactory" ref="connectionFactory" />
    		<property name="defaultDestination" ref="testQueue" />
    		<property name="sessionTransacted" value="false" />
    		
    		<property name="receiveTimeout" value="30000" />
    		  
       		<property name="messageConverter" ref="userMessageConverter"/>  
    	</bean>
    	     
    	<bean id="userMessageConverter" class="cn.slimsmart.activemq.demo.spring.msgconverter.UserMessageConverter"/>  
    </beans>

消费者配置：

<?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:p="http://www.springframework.org/schema/p"
    	xmlns:aop="http://www.springframework.org/schema/aop" xmlns:context="http://www.springframework.org/schema/context"
    	xmlns:jee="http://www.springframework.org/schema/jee" xmlns:jms="http://www.springframework.org/schema/jms"
    	xmlns:tx="http://www.springframework.org/schema/tx" xmlns:util="http://www.springframework.org/schema/util"
    	xmlns:lang="http://www.springframework.org/schema/lang"
    	xsi:schemaLocation="
    			http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-3.0.xsd
    			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://www.springframework.org/schema/jee http://www.springframework.org/schema/jee/spring-jee-3.0.xsd
    			http://www.springframework.org/schema/jms http://www.springframework.org/schema/jms/spring-jms-3.0.xsd
    			http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-3.0.xsd
    			http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util-3.0.xsd
    			http://www.springframework.org/schema/lang http://www.springframework.org/schema/lang/spring-lang-3.0.xsd">
    
    	
    	<bean id="connectionFactory"
    		class="org.springframework.jms.connection.CachingConnectionFactory">
    		
    		<property name="sessionCacheSize" value="10" />
    		
    		<property name="targetConnectionFactory">
    			<bean class="org.apache.activemq.ActiveMQConnectionFactory">
    				
    				<property name="brokerURL" value="tcp://192.168.18.43:61616" />
    			</bean>
    		</property>
    	</bean>
    
    	
    	<bean id="testQueue" class="org.apache.activemq.command.ActiveMQQueue">
    		<property name="physicalName" value="spring.queue.converte" />
    	</bean>
    
    	
    	<bean id="queueContainer"
    		class="org.springframework.jms.listener.DefaultMessageListenerContainer">
    		<property name="connectionFactory" ref="connectionFactory" />
    		<property name="destination" ref="testQueue" />
    		
    		<property name="messageListener" ref="messageListenerAdapter" />
    		<property name="receiveTimeout" value="30000" />
    	</bean>
    
    	
    	<bean id="messageListenerAdapter"
    		class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
    		<property name="delegate" ref="userMessageListener" />
    		<property name="defaultListenerMethod" value="receiveMessage" />
    		<property name="messageConverter" ref="userMessageConverter" />
    	</bean>
    
    	<bean id="userMessageListener"
    		class="cn.slimsmart.activemq.demo.spring.msgconverter.UserMessageListener"></bean>
    	
    	<bean id="userMessageConverter"
    		class="cn.slimsmart.activemq.demo.spring.msgconverter.UserMessageConverter" />
    </beans>

发送消息：

package cn.slimsmart.activemq.demo.spring.msgconverter;
    
    import java.util.Date;
    
    import org.springframework.context.ApplicationContext;
    import org.springframework.context.support.ClassPathXmlApplicationContext;
    import org.springframework.jms.core.JmsTemplate;
    
    @SuppressWarnings("resource")
    public class ProducerMain {
    
    	public static void main(String[] args) {
    		ApplicationContext context = new ClassPathXmlApplicationContext("classpath:msgconverter/producer.xml");
    	    JmsTemplate jmsTemplate = context.getBean(JmsTemplate.class);
    	    User user = new User();
    	    user.setId("1001");
    	    user.setName("jack");
    	    user.setAge(22);
    	    user.setBirthDate(new Date());
    	    jmsTemplate.convertAndSend(user);
    	}
    }

运行消费者 ：

package cn.slimsmart.activemq.demo.spring.msgconverter;
    
    import org.springframework.context.support.ClassPathXmlApplicationContext;
    
    public class ConsumerMain {
    
    	@SuppressWarnings("resource")
    	public static void main(String[] args) {
    	      new ClassPathXmlApplicationContext("classpath:msgconverter/consumer.xml");
    	}
    }

参考文章：

[1.使用spring + ActiveMQ 总结][1._spring _ ActiveMQ]

[1._spring _ ActiveMQ]: http://my.oschina.net/u/1411333/blog/318543