`

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>
 
1
0
分享到:
评论

相关推荐

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

    **ActiveMQ与Spring集成实例——使用消息转换器** 在企业级应用开发中,消息队列(Message Queue,MQ)作为一种解耦和异步处理的重要工具,被广泛应用。Apache ActiveMQ 是一个开源的消息中间件,它支持多种消息...

    activemq与spring整合源代码

    Spring还提供了丰富的模块,如数据访问、Web、测试等,其中Spring JMS模块专门用于集成消息中间件,使得与ActiveMQ的整合变得简单。 三、ActiveMQ与Spring的整合 1. 添加依赖:首先,在项目中引入ActiveMQ和Spring...

    activemq整合spring完整实例代码(内含所有相关jar包)

    将ActiveMQ与Spring整合,可以方便地在Spring应用中使用消息队列,简化配置,并提供事务性消息支持。 这个实例代码是将ActiveMQ与Spring进行整合的一个实际项目,可以在Tomcat服务器上运行。由于提供了所有必要的...

    Spring和ActiveMQ整合的完整实例

    在IT行业中,Spring框架是Java应用开发中的一个关键组件,它提供了一个全面的编程和配置模型,用于构建灵活、可维护的...这个完整的实例是一个很好的学习资源,可以帮助开发者更好地理解和实践Spring与ActiveMQ的整合。

    activemq+spring实例

    Apache ActiveMQ是业界广泛使用的开源消息代理,而Spring框架则提供了与ActiveMQ集成的强大支持。本实例将探讨如何在Spring环境中配置和使用ActiveMQ,包括Queue和Topic两种通信模式,以及同步和异步的消息接收方式...

    ActiveMQ与spring整合封装

    同时,可以使用Spring的`DefaultJmsListenerContainerFactory`配置监听器容器,动态生成实例,以适应多线程环境。 ```xml &lt;bean id="pooledConnectionFactory" class="org.apache.activemq.pool....

    ActiveMQ-P2P文本消息+Spring和ActiveMQ的整合实例源码

    这个注解可以直接在方法上使用,标记该方法为一个消息监听器,当队列中有消息时,Spring 将自动调用该方法处理消息。 例如: ```java @Service public class MyMessageConsumer { @JmsListener(destination = ...

    ActiveMQ与Spring联合使用Demo

    4. **创建消息消费者**:Spring提供了两种方式来消费消息——基于监听器容器的`DefaultMessageListenerContainer`和基于注解的`@JmsListener`。`@JmsListener`更简洁,可以直接在方法上声明: ```java @...

    ActiveMQ与Spring整合示例Demo

    将ActiveMQ与Spring整合,可以方便地在Spring应用中使用消息队列,实现异步处理和分布式通信。 **一、ActiveMQ安装与配置** 在Linux系统中安装ActiveMQ,首先需要下载ActiveMQ的二进制包,解压后将其放置在合适的...

    activemq整合spring

    标题中的“activemq整合spring”指的是在Java环境中,如何将Apache ActiveMQ,一个流行的开源消息代理和消息中间件,与Spring框架集成,以便利用Spring的便利性来管理ActiveMQ的配置和操作。ActiveMQ提供了发布/订阅...

    Spring集成ActiveMQ配置

    5. **发送和接收消息**:在Spring Bean中,使用JmsTemplate发送消息到队列或主题,同时定义监听器接收消息。例如,发送消息: ```java jmsTemplate.convertAndSend(destination, "Hello, ActiveMQ!"); ``` 接收...

    Spring和ActiveMQ的整合实例源码

    9. **Spring注解**:在提供的实例中,可能包含了使用注解的方式配置Spring与ActiveMQ的集成,如`@EnableJms`启动JMS支持,`@JmsListener`定义消息监听器等。 10. **Tomcat服务器**:Tomcat是一个流行的Java Web...

    spring使用activeMQ实现消息发送

    本文将深入探讨如何在Spring环境中使用ActiveMQ来实现消息的发送与接收,以提高系统的可扩展性和解耦性。 首先,我们需要了解Spring对ActiveMQ的支持。Spring提供了`spring-jms`模块,它包含了一组丰富的API和配置...

    activemq与spring整合

    #### 3.4 定义消息监听器 使用 Spring AOP(面向切面编程)或 `DefaultMessageListenerContainer` 来监听消息队列。例如,创建一个 `MessageListener` 接口的实现类,并通过 `@JmsListener` 注解来指定监听的队列或...

    activemq+spring demo 简单示例222

    Spring与ActiveMQ的整合,使得我们可以方便地在Spring应用中使用消息队列服务。 整合步骤如下: 1. **安装ActiveMQ**:首先,我们需要下载并安装ActiveMQ。可以从官方网站获取最新版本,解压后在bin目录下启动...

    spring 与ACTIVEMQ整合

    3. **Spring配置**:在Spring应用中整合ActiveMQ,首先需要在配置文件中引入ActiveMQ的JAR包,并配置连接工厂、目的地(队列或主题)以及消息监听器容器。 二、Spring整合ActiveMQ步骤 1. **添加依赖**:在项目中...

    Spring 使用ActiveMQ 收发消息实例

    Spring与ActiveMQ的集成使得在分布式系统中实现可靠的消息通信变得简单高效。ActiveMQ是Apache软件基金会开发的一个开源消息中间件,它遵循Java消息服务(JMS)规范,提供了多种协议支持,如AMQP、STOMP等。 首先,...

    ActiveMQ+Spring完整详解例子

    3. **Spring与ActiveMQ的集成** - 引入Spring的JMS模块依赖,如`spring-jms`。 - 创建`ConnectionFactory`,配置连接ActiveMQ的URL、用户名和密码。 - 使用`&lt;jee:jndi-lookup&gt;`标签或`@Resource`注解注入`...

Global site tag (gtag.js) - Google Analytics