ActiveMQ与spring集成实例之使用消息监听器

    在EJB世界里，JMS消息最常用的功能之一是用于实现消息驱动Bean（MDB）。Spring提供了一个方法来创建消息驱动的POJO（MDP），并且不会把用户绑定在某个EJB容器上。

   通常用消息监听器容器从JMS消息队列接收消息并驱动被注射进来的MDP。消息监听器容器负责消息接收的多线程处理并分发到各MDP中。一个消息侦听容器是MDP和消息提供者之间的一个中介，用来处理消息接收的注册，事务管理的参与，资源获取和释放，异常转换等等。这使得应用开发人员可以专注于开发和接收消息（可能的响应）相关的（复杂）业务逻辑，把和JMS基础框架有关的样板化的部分委托给框架处理。 

Spring提供了三种 AbstractMessageListenerContainer 的子类，每种各有其特点。
 
第一种：SimpleMessageListenerContainer

   这个消息侦听容器是三种中最简单的。它在启动时创建固定数量的JMS session并在容器的整个生命周期中使用它们。这个类不能动态的适应运行时的要求或参与消息接收的事务处理。然而它对JMS提供者的要求也最低。它只需要简单的JMS API。

第二种：DefaultMessageListenerContainer

   这个消息侦听器使用的最多。和 SimpleMessageListenerContainer 相反，这个子类可以动态适应运行时侯的要求，也可以参与事务管理。每个收到的消息都注册到一个XA事务中（如果使用 JtaTransactionManager 配置过），这样就可以利用XA事务语义的优势了。这个类在对JMS提供者的低要求和提供包括事务参于等的强大功能上取得了很好的平衡。

第三种：ServerSessionMessageListenerContainer

   这个监听器容器利用JMS ServerSessionPool SPI动态管理JMS Session。 使用者各种消息监听器可以获得运行时动态调优功能，但是这也要求JMS提供者支持ServerSessionPool SPI。如果不需要运行时性能调整，请使用 DefaultMessageListenerContainer 或 SimpleMessageListenerContainer。 

 

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

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

import com.acca.activemq.entity.User;

/**
 * 
 * 自定义消息侦听器
 * 
 * @author zhouhua, 2012-12-26
 */
public class MyMessageListener implements MessageListener {

    /**
     * @param arg0
     * @see javax.jms.MessageListener#onMessage(javax.jms.Message)
     */
    @Override
    public void onMessage(Message message) {
       System.out.println(message.toString());
       System.out.println("MyMessageListener");
       if(message instanceof ObjectMessage){
           ObjectMessage objectMessage=(ObjectMessage) message;
           try {
            User user=(User) objectMessage.getObject();
            System.out.println(user.getUserName());
        } catch (JMSException e) {
            e.printStackTrace();
        }
       }
    }

}

 

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

import java.io.Serializable;

import javax.jms.JMSException;
import javax.jms.Message;
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((Serializable) object);
        return msg; 
    }

}

 

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

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

import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;

import com.acca.activemq.entity.User;

/**
 * 
 * 
 *
 * @author zhouhua, 2012-12-26
 */
public class Sender {
    
    public static void main(String[] args) {
        
        ApplicationContext applicationContext = new ClassPathXmlApplicationContext("applicationContext.xml");

        JmsTemplate template = (JmsTemplate) applicationContext.getBean("jmsTemplate");

        Destination destination = (Destination) applicationContext.getBean("destination");
        User user=new User();
        user.setUserName("张三");
        template.convertAndSend(user);
        System.out.println("成功发送了一条JMS消息");

    }
}

 

/*
 * 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-26
 */
public class User implements Serializable{

    private String userName;

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

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

}

 

<?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: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/context 
        http://www.springframework.org/schema/context/spring-context-2.5.xsd"
	default-autowire="byName">


	<!-- 配置connectionFactory -->
	<bean id="jmsFactory" class="org.apache.activemq.pool.PooledConnectionFactory"
		destroy-method="stop">
		<property name="connectionFactory">
			<bean class="org.apache.activemq.ActiveMQConnectionFactory">
				<property name="brokerURL">
					<value>tcp://127.0.0.1:61616</value>
				</property>
			</bean>
		</property>
		<property name="maxConnections" value="100"></property>
	</bean>

	<!-- Spring JMS Template -->
	<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
		<property name="connectionFactory">
			<ref local="jmsFactory" />
		</property>
		<property name="defaultDestinationName" value="subject" />
		<!-- 区别它采用的模式为false是p2p，为true是订阅 -->
		<property name="pubSubDomain" value="true" />
		<property name="messageConverter" ref="myConvert"></property>
	</bean>

	<!-- 发送消息的目的地（一个队列） -->
	<bean id="destination" class="org.apache.activemq.command.ActiveMQTopic">
		<!-- 设置消息队列的名字 -->
		<constructor-arg index="0" value="subject" />
	</bean>
	
    <!-- 消息监听  -->  
    <bean id="listenerContainer"  
        class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <!-- 消息监听器输出消息的数量 -->  
        <property name="concurrentConsumers" value="1" />  
        <property name="connectionFactory" ref="jmsFactory" />  
        <property name="destinationName" value="subject" />  
        <property name="messageListener" ref="myMessageListener" />  
        <property name="pubSubNoLocal" value="false"></property>  
    </bean> 
     
	<!-- 消息转换器 -->
	<bean id="myConvert" class="com.acca.activemq.convert.MyConvert">
	</bean>

   <!-- 消息侦听器 -->
   <bean id="myMessageListener" class="com.acca.activemq.listener.MyMessageListener"></bean>

   <!-- 
	<bean id="messageReceiver" class="com.acca.activemq.ProxyJMSConsumer">
		<property name="jmsTemplate" ref="jmsTemplate"></property>
		<property name="destination" ref="destination"></property>
	</bean>
	-->

    

</beans>