当柳上原的风吹向天际的时候...

真正的快乐来源于创造
  BlogJava :: 首页 :: 联系 :: 聚合  :: 管理
  368 Posts :: 1 Stories :: 201 Comments :: 0 Trackbacks

公告

常用链接

留言簿(3)

随笔分类

相册

个人常用链接

最新随笔

积分与排名

  • 积分 - 392402
  • 排名 - 145

最新评论

阅读排行榜
在ActiveMQ5.3.0上建立队列和主题以及通过它们收发消息的实现
相对于Weblogic这种大家伙来说，在ActiveMQ5.3.0上建立队列和主题还是相对容易的，当然如果路没走对也要费一番周折，本人在此写出小文，希望能对ActiveMQ初学者有所裨益，顺便感谢网络上那些辛勤的有共享精神的ActiveMQ研究者和鄙视一下那些懒惰的有剽窃习惯的转载者。

一.下载安装ActiveMQ。
在http://activemq.apache.org/ 你可以得到ActiveMQ的最新版本，本人下载的是win版本，下载到本地后解压到一个目录（以后简称ActiveMQ解压目录）就算安装完毕了，确实很简单。

二.启动ActiveMQ。
启动它也很容易，找到ActiveMQ解压目录下的bin目录，双击activemq.bat就可以了，如果嫌麻烦可以做成一个批处理文件，然后让计划任务自动调用它。activemq.bat启动后会出现一个命令行窗口，这意味着broker已经准备就绪了。

三.创建队列和主题
ActiveMQ启动后，我们可以就使用 http://localhost:8161/admin 进入ActiveMQ的消息管理后台系统，如下图：

在这个页面中有一排横向菜单，如果你需要创建队列Queue，点击Queues菜单，输入你想要的队列的名称，再点击Create按钮，队列即被创建好了；主题也很类似，你点击Topics菜单，输入你想要的主题的名称，再点击create按钮，主题也就建立好了。确实是非常快捷方便。本人建立了队列DealQueue和主题BuyTopic，用于下面的程序。

四.利用队列和主题收发消息。
有了Spring的帮助，JMS处理能省却很多功夫，我们只需要进行对应的配置即可，下面是通过队列DealQueue和主题BuyTopic收发消息的配置和消息发送器监听器源代码，大家可以参考，另外也可以从这 这里（注意为了减小体积，包都被删除了，请到ActiveMQ安装目录的lib寻找相应的包载入，其它通用包请自行寻找）    下载源码查看。

Spring上下文配置文件bean.xml，这里文件里配置的有连接工厂，jms模板，目的地，消息发送者和接受者等。
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE beans PUBLIC "-//SPRING//DTD BEAN//EN" "http://www.springframework.org/dtd/spring-beans.dtd">

<beans>
    <!-- 连接工厂 -->
    <bean id="connectionFactory"
        class="org.apache.activemq.ActiveMQConnectionFactory">
        <property name="brokerURL">
            <value>tcp://192.168.0.1:61616</value>
        </property>

    </bean>

    <!-- JMS模板 -->
    <bean id="jmsTemplate"
        class="org.springframework.jms.core.JmsTemplate">
        <property name="connectionFactory">
            <ref bean="connectionFactory" />
        </property>
    </bean>

    <!-- **************************** -->
    <!--     以下是队列DealQueue的配置   -->
    <!-- **************************** -->
    <!-- 在ActiveMQ上配置的队列DealQueue -->
    <bean id="dest4DealQueue"
        class="org.apache.activemq.command.ActiveMQQueue">
        <constructor-arg index="0">
            <value>DealQueue</value>
        </constructor-arg>
    </bean>
    
    <!-- JMS 消息发送者 ,用于给DealQueue发送消息 -->
    <bean id="jmsSender4DealQueue"
        class="com.heyang.sender.JmsSender">
        <property name="jmsTemplate" ref="jmsTemplate" />
        <property name="destination" ref="dest4DealQueue" />
    </bean> 
    
    <!-- JMS 消息监听容器,Spring容器启动后自行监听DealQueue,有消息后会交给dealQueueListener处理 -->
    <bean id="listenerContainer4DealQueue"
        class="org.springframework.jms.listener.DefaultMessageListenerContainer">        
        <property name="connectionFactory" ref="connectionFactory" />
        <property name="concurrentConsumers" value="1" />
        <property name="destination" ref="dest4DealQueue" />
        <property name="messageListener" ref="dealQueueListener" />
    </bean>
    
    <!-- MDP:用于接收DealQueue上的消息 -->
    <bean id="dealQueueListener" class="com.heyang.listener.DealQueueListener" />
    
    <!-- **************************** -->
    <!--     以下是主题BuyTopic的配置   -->
    <!-- **************************** -->
    <!-- 在ActiveMQ上配置的主题 -->
    <bean id="dest4BuyTopic"
        class="org.apache.activemq.command.ActiveMQTopic">
        <constructor-arg index="0">
            <value>BuyTopic</value>
        </constructor-arg>
    </bean>
    
    <!-- JMS 消息发送者 ,用于给BuyTopic发送消息 -->
    <bean id="jmsSender4BuyTopic"
        class="com.heyang.sender.JmsSender">
        <property name="jmsTemplate" ref="jmsTemplate" />
        <property name="destination" ref="dest4BuyTopic" />
    </bean> 
    
    <!-- JMS 消息监听容器,Spring容器启动后自行监听BuyTopic,有消息后会交给buyTopicListener处理 -->
    <bean id="listenerContainer4BuyTopic"
        class="org.springframework.jms.listener.DefaultMessageListenerContainer">        
        <property name="connectionFactory" ref="connectionFactory" />
        <property name="concurrentConsumers" value="1" />
        <property name="destination" ref="dest4BuyTopic" />
        <property name="messageListener" ref="buyTopicListener" />
    </bean>
    
    <!-- MDP:用于接收BuyTopic上的消息 -->
    <bean id="buyTopicListener" class="com.heyang.listener.BuyTopicListener" />
</beans>


三个对应的类：
1.JMS消息发送者：
package com.heyang.sender;

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

import org.apache.activemq.Message;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;

/**
 * 此类用发送消息
 * @author 何杨
 * @date 2009-10-15
 * @time 上午07:44:15
 */
public class JmsSender {

    // JMS模板    
    private JmsTemplate jmsTemplate;

    // 目的地
    private Destination destination;

    public void send(final String msg) {
        jmsTemplate.send(destination, new MessageCreator() {
            public Message createMessage(Session session) throws JMSException {
                return (Message) session.createTextMessage(msg);
            }
        });
    }

    public JmsTemplate getJmsTemplate() {
        return jmsTemplate;
    }

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

    public Destination getDestination() {
        return destination;
    }

    public void setDestination(Destination destination) {
        this.destination = destination;
    }
}

JMS消息监听者，用于监听BuyTopic：
package com.heyang.listener;

import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;

import org.apache.log4j.Logger;

/**
 * JMS消息监听者，用于监听BuyTopic。
 * 
 * @author 何杨
 * @date 2009-10-13
 * @time 下午04:10:33
 */
public class BuyTopicListener implements MessageListener {
    public static Logger logger = Logger.getLogger(DealQueueListener.class);
    
    public void onMessage(Message message) {
        if (message instanceof TextMessage) {            
           
            // 消息转化
            String msg="";
            try{
                msg=((TextMessage) message).getText();
                System.out.println("从BuyTopic收到消息:"+msg);
            } catch (Exception ex) {
                logger.error(ex);
                ex.printStackTrace();
                return;
            }
        }
    }
}


JMS消息监听者，用于监听DealQueue：
package com.heyang.listener;

import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;

import org.apache.log4j.Logger;

/**
 * JMS消息监听者，用于监听DealQueue。
 * 
 * @author 何杨
 * @date 2009-10-13
 * @time 下午04:10:33
 */
public class DealQueueListener implements MessageListener {
    public static Logger logger = Logger.getLogger(DealQueueListener.class);
    
    public void onMessage(Message message) {
        if (message instanceof TextMessage) {            
           
            // 消息转化
            String msg="";
            try{
                msg=((TextMessage) message).getText();
                System.out.println("从DealQueue收到消息:"+msg);
            } catch (Exception ex) {
                logger.error(ex);
                ex.printStackTrace();
                return;
            }
        }
    }
}
    

用户可以使用Test.java中的语句来测试一下效果。另外有一点想赘述一下，上述两个消息监听者是受Spring容器控制的，它们之所以能成为Message Diven Pojo是因为Spring容器自启动开始就对目标进行轮询，如果发现有消息就通知指定消息监听者响应，这样消息监听者的onMessage函数就会被调用，其背后的调用者即上面配置文件中的listenerContainer4DealQueue和listenerContainer4BuyTopic，我们虽没有直接使用它们但它们从Spring上下文启动伊始就一直开始工作了，从log4j的输出就能看出这一点，当然为了消除轮询输出的消极影响，你可以调高log4j的输出级别。

但愿此文能对你有所帮助，全文完。
posted on 2009-10-27 09:00 何杨 阅读(4134) 评论(0)  编辑  收藏