`

结合Spring2.0和ActiveMQ进行异步消息调用_此文为转载

    博客分类：
  • jms
jms 
阅读更多
在Spring 2.0之前，Spring的JMS的作用局限于产生消息。这个功能（封装在 JmsTemplate 类中）当然是很好的，但是，它没有描述完整的JMS堆栈，比如像消息的 异步 产生和消耗。JMS堆栈缺少的这一部分已经被添加，Spring 2.0现在提供对消息异步消耗的完整支持。
      让我们从一个例子开始。
      首先我们打开ActiveMQ。从ActiveMQ的安装路径上的bin目录，那里有一个ActiveMQ.bat，双击执行即可。不过要注意必须先设置java_home环境变量。ActiveMQ默认的服务端口是61616。
      然后我们开始配置Spring配置文件。我起名为spring-jms.xml


首先要配置一个ConnectionFactory代码如下

Copy code

<bean id="connectionFactory"
        class="org.apache.activemq.ActiveMQConnectionFactory">
        <property name="brokerURL" value="tcp://localhost:61616" />
</bean>

 
这里用到的ConnectionFactory是ActiveMQ提供的工厂，为了能使用这个工厂，我们必须在项目中添加以下几个jar文件：
geronimo-jms_1.1_spec-1.0.jar,
activeio-core-3.0-beta3.jar,
activemq-core-4.0.1.jar,
backport-util-concurrent-2.1.jar,
commons-logging-1.0.4.jar,
geronimo-j2ee-management_1.0_spec-1.0.jar
以上这些Jar文件都存在于ActiveMQ安装目录的lib目录下，这些可是我一个一个试验出来的，累个半死。。


然后应该配置一个Queue（我使用的是点对点方式），不过ActiveMQ只要提供一个名字就可以自动创建队列，因此这一步省了，呵呵

下 面就轮到Spring的支持类了，首先是JmsTemplate。这个类提供了大量的方法简化我们对JMS的操作。常用的有两个， org.springframework.jms.core.JmsTemplate102和 org.springframework.jms.core.JmsTemplate，这两个类分别支持JMS的1.02版本和1.1版本。现在比较常用 的还是1.02版本。配置如下

Copy code


    <bean id="jmsTemplate"
        class="org.springframework.jms.core.JmsTemplate102">
        <property name="connectionFactory" ref="connectionFactory"/>
        <property name="timeToLive" value="86400000"/>
        <property name="defaultDestinationName" value="cmpp" />
        <property name="messageConverter" ref="messageConverter" />
        <property name="receiveTimeout" value="30000" />
    </bean>


上 面的配置中用到了第一步配置的connectionFactory以及一个消息转换的类messageConverter，这个类实现了 org.springframework.jms.support.converter.MessageConverter接口，可以在消息发送之前和接 受之后进行消息类型转换。具体的看最后的实例代码。配置代码如下：

Copy code
    <!-- Spring JMS SimpleConverter -->
       <bean id="simpleConverter"  class="org.springframework.jms.support.converter.SimpleMessageConverter" />
<!-- Message Converter -->
    <bean id="messageConverter"
        class="com.liangj.apmgt.jms.ApmgtMessageConverter">
        <property name="converter">
            <ref local="simpleConverter" />
        </property>
    </bean>


这里还配置了发送的消息的存在时间timeToLive，目标Queue的名字defaultDestinationName，接受消息超时时间receiveTimeout

配置发送代码

Copy code
    <!-- Message porducer -->
    <bean id="producer"
        class="com.liangj.apmgt.jms.DefaultApmgtMessageProducer">
        <property name="jmsTemplate" ref="jmsTemplate" />
    </bean>



接着配置监听器，这是Spring2.0新增的功能，配置如下：

Copy code

    <!-- this is the Message Driven POJO (MDP) -->
    <bean id="messageListener"
        class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
        <constructor-arg>
            <bean
                class="com.liangj.apmgt.jms.DefaultApmgtMessageListener" />
        </constructor-arg>
        <property name="defaultListenerMethod" value="onMessage" />
        <property name="messageConverter" ref="messageConverter" />
    </bean>

    <!-- and this is the attendant message listener container -->
    <bean id="listenerContainer"
        class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="connectionFactory" />
        <property name="destinationName" value="${jms.destinationName.cmpp}" />
        <property name="messageSelector" value="${jms.messageSelector}" />
        <property name="messageListener" ref="messageListener" />
    </bean>




      Spring配置监听器有很多种选择，在这里我选择这回种MessageListenerAdapter方法主要是因为这个方法比较灵活。实现他只要一个很普通的java类即可，和JMS以及Spring的耦合度最低。其中方法onMessage可以随便修改方法名，只要在配置文件中对应的修改就好了。
     MessageListenerAdapter还有一个功能就是如果处理方法(我这里是onMessage)返回一个非空值，它将自动返回一个响应消息。这个消息会返回给JMS Reply-To属性定义的目的地(如果存在)，或者是MessageListenerAdapter设置(如果配置了)的缺省目的地；如果没有定义目的地，那么将产生一个InvalidDestinationException异常(此异常将不会只被捕获而不处理，它将沿着调用堆栈上传)。
     这样我们的配置就都完成了。接下来我们来实现对应的Java文件
先是接口文件发送消息接口IApmgtMessageProducer.java

Copy code

public interface IApmgtMessageProducer {
    public abstract void sendMessage(ApmgtMessageData messageData);
}


接受消息接口IApmgtMessageListener.java

Copy code
public interface IApmgtMessageListener {
    public void onMessage(ApmgtMessageData message);
}


发消息的文件DefaultApmgtMessageProducer.java

Copy code
public class DefaultApmgtMessageProducer implements IApmgtMessageProducer {
   
    private JmsTemplate jmsTemplate;

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

    public void sendMessage(ApmgtMessageData messageData) {
        this.jmsTemplate.convertAndSend(messageData);
    }
}


收消息文件DefaultApmgtMessageListener.java

Copy code

public class DefaultApmgtMessageListener implements IApmgtMessageListener {
    public void onMessage(ApmgtMessageData message) {
        System.out.println("监听到消息："+message);
    }
}


消息转换类ApmgtMessageConverter.java

Copy code
public class ApmgtMessageConverter implements MessageConverter {

    private Log log = LogFactory.getLog(ApmgtMessageConverter.class);

    private SimpleMessageConverter converter;

    public void setConverter(SimpleMessageConverter converter) {
        this.converter = converter;
    }

    public Object fromMessage(Message message) throws JMSException, MessageConversionException {
        if (message instanceof ObjectMessage) {
            ObjectMessage o_message = (ObjectMessage)message;
            MessageHeader header = new MessageHeader();
            header.setId(message.getLongProperty("id"));
            header.setReceiver(message.getIntProperty("receiver"));
            header.setSender(message.getIntProperty("sender"));
            header.setSendPerson(message.getStringProperty("sendPerson"));
            header.setType(message.getIntProperty("type"));
            Serializable messageContent = o_message.getObject();
            ApmgtMessageData<Serializable> messageData = new ApmgtMessageData<Serializable>();
            messageData.setMessageContent(messageContent);
            messageData.setMessageHeader(header);
            return messageData;
        }
        return null;
    }

    public Message toMessage(Object object, Session session) throws JMSException, MessageConversionException {
        if (object instanceof ApmgtMessageData) {
            ApmgtMessageData data = (ApmgtMessageData) object;
            Message message = converter.toMessage(data.getMessageContent(), session);
            message.setLongProperty("id", data.getMessageHeader().getId());
            message.setIntProperty("receiver", data.getMessageHeader().getReceiver());
            message.setIntProperty("sender", data.getMessageHeader().getSender());
            message.setIntProperty("type", data.getMessageHeader().getType());
            message.setStringProperty("sendPerson", data.getMessageHeader().getSendPerson());
            log.info("发送消息[MessageSender]:\n" + message);
             return message;
        } else {
            return null;
        }
    }

}


消息类文件  消息父类：ApmgtMessageData.java

Copy code
public class ApmgtMessageData<T extends Serializable>{

    protected T messageContent;
   
    protected MessageHeader messageHeader;

    public T getMessageContent() {
        return this.messageContent;
    }

    public MessageHeader getMessageHeader() {
        return this.messageHeader;
    }

    public void setMessageContent(T messageContent) {
        this.messageContent = messageContent;
    }
    public void setMessageHeader(MessageHeader messageHeader) {
        this.messageHeader = messageHeader;
    }

}



消息属性的一个类MessageHeader.java

Copy code
public class MessageHeader {

    /**
     * 消息ID
     */
    private long id;

    /**
     * 消息类型
     */
    private int type;

    /**
     * 消息发送方，发送消息的模块
     */
    private int sender;

    /**
     * 消息接收方，接收消息的模块
     */
    private int receiver;

    /**
     * 消息发送者，具体的用户
     */
    private String sendPerson;

    public MessageHeader(){
        this.id = System.currentTimeMillis() ;
    }
   
    public long getId() {
        return id;
    }

    public void setId(long id) {
        this.id = id;
    }

    public String getSendPerson() {
        return sendPerson;
    }

    public void setSendPerson(String sendPerson) {
        this.sendPerson = sendPerson;
    }

    public int getReceiver() {
        return receiver;
    }

    public void setReceiver(int receiver) {
        this.receiver = receiver;
    }

    public int getSender() {
        return sender;
    }

    public void setSender(int sender) {
        this.sender = sender;
    }

    public int getType() {
        return type;
    }

    public void setType(int type) {
        this.type = type;
    }

}



消息体的类ModPasswordRequest.java

Copy code
public class ModPasswordRequest implements Serializable{


    private static final long serialVersionUID = 1L;

    /**
     * 旧密码
     */
    private String oldPassword;
   
    /**
     * 新密码
     */
    private String newPassword;

    public String getNewPassword() {
        return newPassword;
    }

    public void setNewPassword(String newPassword) {
        this.newPassword = newPassword;
    }

    public String getOldPassword() {
        return oldPassword;
    }

    public void setOldPassword(String oldPassword) {
        this.oldPassword = oldPassword;
    }

}



消息类：ApmgtModPasswordRequest.java

Copy code
public class ApmgtModPasswordRequest extends ApmgtMessageData<ModPasswordRequest> {
   
   
   
    private static final int REQ_MODPASSWORD = 0;
    private static final int INTF = 1;
    private static final int APMGT = 2;

    public void init(){
        messageHeader = new MessageHeader();
        messageContent = new ModPasswordRequest();
        messageHeader.setType(REQ_MODPASSWORD);
        messageHeader.setSender(INTF);
        messageHeader.setReceiver(APMGT);
        messageContent.setNewPassword("123456");
        messageContent.setOldPassword("654321");
    }
 
}



最后是测试类Main.java

Copy code
    public class Main {

    public static void main(final String[] args) throws Exception {
       
        PropertyConfigurator.configure("log4j.properties");
       
        AbstractApplicationContext ctx = new ClassPathXmlApplicationContext(new String[] { "spring-jms.xml" });
//        ctx.registerShutdownHook();
       
        IApmgtMessageProducer producer = (IApmgtMessageProducer)ctx.getBean("producer");
       
        ApmgtModPasswordRequest messageData = new ApmgtModPasswordRequest();
        messageData.setMessageHeader(new MessageHeader());
        messageData.setMessageContent(new ModPasswordRequest());
        messageData.init();
       
        producer.sendMessage(messageData);
    }
}



还有两个配置文件，第一个spring-jms.xml

Copy code
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="
    xmlns:xsi="
    xsi:schemaLocation="http://www.springframework.org/schema/beans


    <bean id="propertyConfigurer"
        class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
        <property name="locations">
            <list>
                <value>apmgt.properties</value>
            </list>
        </property>
    </bean>

    <!-- ####################################### -->
    <!--          JMS Spring Beans               -->
    <!-- ####################################### -->

    <!-- Jms ConnectionFactory -->
    <bean id="connectionFactory"
        class="org.apache.activemq.ActiveMQConnectionFactory">
        <property name="brokerURL" value="${jms.brokerURL}" />
    </bean>

    <!-- Spring JMS SimpleConverter -->
    <bean id="simpleConverter"
        class="org.springframework.jms.support.converter.SimpleMessageConverter" />

    <!-- JMS Queue Template -->
    <bean id="jmsTemplate"
        class="org.springframework.jms.core.JmsTemplate102">
        <property name="connectionFactory" ref="connectionFactory"/>
        <property name="timeToLive" value="${jms.timeToLive}"/>
        <property name="defaultDestinationName" value="${jms.destinationName.cmpp}" />
        <property name="messageConverter" ref="messageConverter" />
        <property name="receiveTimeout" value="${jms.receiveTimeout}" />
    </bean>

    <!-- Message Converter -->
    <bean id="messageConverter"
        class="com.liangj.apmgt.jms.ApmgtMessageConverter">
        <property name="converter">
            <ref local="simpleConverter" />
        </property>
    </bean>

    <!-- Message porducer -->
    <bean id="producer"
        class="com.liangj.apmgt.jms.DefaultApmgtMessageProducer">
        <property name="jmsTemplate" ref="jmsTemplate" />
    </bean>

    <!-- this is the Message Driven POJO (MDP) -->
    <bean id="messageListener"
        class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
        <constructor-arg>
            <bean
                class="com.liangj.apmgt.jms.DefaultApmgtMessageListener" />
        </constructor-arg>
        <property name="defaultListenerMethod" value="onMessage" />
        <property name="messageConverter" ref="messageConverter" />
    </bean>

    <!-- and this is the attendant message listener container -->
    <bean id="listenerContainer"
        class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="connectionFactory" />
        <property name="destinationName" value="${jms.destinationName.cmpp}" />
        <property name="messageSelector" value="${jms.messageSelector}" />
        <property name="messageListener" ref="messageListener" />
    </bean>
</beans>



apmgt.properties

Copy code
#jms properties
jms.brokerURL=tcp://localhost:61616
jms.receiveTimeout=3000
jms.destinationName.cmpp=cmpp
jms.messageSelector=receiver=2
#one day is 86400000 ms. 0 is means that it lives forever.
jms.timeToLive=86400000



最后后还有源代码，希望对大家有帮助，写了2个小时，真累
ActiveMQ和Tomcat结合使用，网上有很多文章了，这里就不说了，over


分享到：
| web系统错误处理
评论
发表评论

您还没有登录,请您登录后再发表评论

相关推荐

    结合Spring2.0和ActiveMQ进行异步消息调用

    总结，结合Spring 2.0和ActiveMQ进行异步消息调用，可以提高系统的响应速度和可扩展性，降低组件间的耦合。通过理解JMS模板、监听容器和ActiveMQ的相关配置，开发者可以构建出高效、可靠的异步消息处理系统。在实际...

    Spring-ActiveMQ.rar_Spring Activemq_activemq_activemq spring

    在IT领域，消息队列（Message Broker）是一个重要的组件，它允许分布式系统中的不同部分通过异步通信来解耦和协调工作。ActiveMQ是Apache软件基金会的一个开源项目，它是Java消息服务（JMS）的一个实现，广泛应用于...

    activemq-spring-1.2_001.jar.zip

    在Java世界中，消息中间件ActiveMQ扮演着至关重要的角色，它提供了高效、可靠的异步通信机制，而Spring框架则以其强大的依赖注入和面向切面编程能力深受开发者喜爱。当这两者结合时，能构建出更健壮、灵活的应用系统...

    Spring和ActiveMQ整合的完整实例

    而ActiveMQ则是Apache软件基金会的一个开源项目，它是Java消息服务（JMS）的实现，允许应用程序通过消息传递进行异步通信。将Spring与ActiveMQ整合，可以轻松地在Spring应用中实现消息队列的功能，提高系统的可扩展...

    spring使用activeMQ实现消息发送

    综上所述，Spring与ActiveMQ的结合使用能帮助开发者构建高效、可靠的分布式系统，通过消息队列进行异步通信，提高了系统的并发能力和容错性。在实际项目中，可以根据具体需求调整配置和代码，实现复杂的消息处理场景...

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

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

    Spring+ActiveMQ消息队列+前台接收消息

    6. **事务管理**：Spring和ActiveMQ可以结合使用JTA事务，确保消息发送和业务操作的原子性。当开启JTA事务时，如果业务操作失败，消息也会被回滚，不会被发送到队列。 **前台接收消息** 在Web应用中，前台接收消息...

    spring整合Activemq源码

    总结，`spring整合Activemq源码`项目展示了如何在Spring和SpringMVC环境中利用ActiveMQ实现消息的发布与订阅。通过理解这些知识点，开发者能够更好地理解和实践消息队列在实际项目中的应用，提升系统的稳定性和扩展...

    Spring集成ActiveMQ配置

    Spring集成ActiveMQ是将Spring框架与ActiveMQ消息中间件相结合，实现异步处理和解耦应用程序的关键技术。在本文中，我们将深入探讨如何配置和使用这一组合，以及它在实际项目中的应用。 首先，让我们了解Spring框架...

    flex 结合spring activemq 做的一个简易聊天室

    标题中的“flex结合spring activemq做了一个简易聊天室”涉及到的是使用Adobe Flex技术构建前端UI，通过Spring框架与ActiveMQ消息中间件进行通信，实现一个简单的聊天室应用。这个项目是一个很好的示例，展示了如何...

    Spring和ActiveMQ的整合实例源码

    本实例“Spring和ActiveMQ的整合实例源码”旨在展示如何将Spring框架与ActiveMQ结合使用，利用JMS进行高效的消息传递。这个实例适合那些希望学习或深入了解Spring框架如何与消息中间件集成的开发者。它基于Spring的...

    spring2 activemq5 tomcat6构建jms

    标题“spring2 activemq5 tomcat6构建jms”涉及了几个关键的Java技术，主要集中在企业级应用开发中的消息传递系统。首先，我们来深入理解这些技术及其相互关系。 Spring框架是Java应用程序开发的一个核心工具，尤其...

    多个spring 和ActiveMQ结合的项目

    而ActiveMQ则是Apache软件基金会的一个开源消息代理，它实现了多种消息协议，包括JMS（Java Message Service），用于在分布式系统中进行异步通信。当我们谈到"多个Spring和ActiveMQ结合的项目"时，这意味着我们正在...

    Spring boot 和内置ActiveMQ集成例子.zip

    而Apache ActiveMQ是流行的开源消息代理，符合Java Message Service（JMS）标准，用于处理异步通信和消息传递。将Spring Boot与ActiveMQ集成可以提供强大的消息处理能力，使应用能够解耦组件，提高可扩展性和容错性...

    spring-boot-activemq-demo

    通过这个示例，你可以了解到如何在Spring Boot应用中集成ActiveMQ，使用Queue和Topic进行消息传递，并了解连接池和消息确认机制。这些技术在分布式系统、微服务架构中广泛用于实现异步通信和解耦。

    springboot集成activemq实现消息接收demo

    这个简单的Demo展示了如何在Spring Boot中集成ActiveMQ进行消息接收。通过这种方式，你可以构建出一个可靠的、异步处理系统，提高应用的性能和可扩展性。在实际生产环境中，ActiveMQ还可以与其他中间件、微服务架构...

    Spring整合Blazeds实现ActiveMQ JMS消息服务

    标题中的“Spring整合Blazeds实现ActiveMQ JMS消息服务”指的是在Java应用程序中使用Spring框架与Blazeds（一个Flex和Java之间的消息传递中间件）集成，通过ActiveMQ（一个流行的开源JMS提供商）来实现消息队列服务...

    Spring和ActiveMQ的整合实例源码ActiveMQSpringDemo.zip

    本实例"Spring和ActiveMQ的整合实例源码ActiveMQSpringDemo"提供了一个完整的解决方案，帮助开发者了解如何在Spring环境中集成ActiveMQ进行消息队列的使用。 首先，Spring是一个广泛使用的Java企业级应用开发框架，...

    activemq和spring整合发布消息和订阅消息demo

    这个"activemq和spring整合发布消息和订阅消息demo"是用于演示如何在Spring环境中配置和使用ActiveMQ来实现实时的数据通信。现在，我们将深入探讨相关的知识点。 首先，ActiveMQ是Apache软件基金会开发的一个开源...

    activeMQ与spring整合开发的例子程序

    本例子程序的主题是"activeMQ与spring整合开发"，这意味着它将展示如何在Spring应用中集成ActiveMQ，以便利用消息队列进行异步通信和解耦组件。以下是一些关键知识点： 1. **Spring集成ActiveMQ的基本配置**：在...

Magicbox
Global site tag (gtag.js) - Google Analytics