ActiveMq-JMS简单实例使用tomcat

wxyfj

ActiveMq初级操作

（工具：tomcat6.0、activeMq5.3.2、Myeclipse7.0，java1。5）

这是本人第一次发表个人使用ActiveMq，如有错误，请指出。只为与众人共享我的心得。

activeMq是一个开源的支持JMS的框架：（以下为考录他人的信息，如有版权问题，请联系）

一、特性及优势

1、实现JMS1.1规范，支持J2EE1.4以上

2、可运行于任何jvm和大部分web容器（ActiveMQ works great in any JVM）

3、支持多种语言客户端（java, C, C++, AJAX, ACTIONSCRIPT等等）

4、支持多种协议（stomp，openwire，REST）

5、良好的spring支持（ActiveMQ has great Spring Support）

6、速度很快，JBossMQ的十倍（ActiveMQ is very fast; often 10x faster than JBossMQ.）

7、与OpenJMS、JbossMQ等开源jms provider相比，ActiveMQ有Apache的支持，持续发展的优势明显。

二、下载安装

1、下载

http://activemq.apache.org/download.html，在这个页面里有你所需要的各个版本，我用的是最新的版本-ActiveMQ 5.3.2。

2、安装

直接解压至任意目录（如：d:\ apache-activemq-5.1.0），但是我用的时候是解压到D盘的根目录下，我建议各位也这样，不用对它的一些bat进行修改。

3、启动ActiveMQ服务器

方法1：

直接运行bin\activemq.bat

方法2（在JVM中嵌套启动）：（这个我没用过，你可以试试）

cd example

ant embedBroker

4、ActiveMQ消息管理后台系统： http://localhost:8161/admin

打开主页后，你可以看到activeMq中带的一些实例，

三、详细实例

以下为本人的一个简单的实例，希望对各位有所帮助，

我是利用网上的资料，将tomcat和activeMq整合在一起做的一个发送接受的发布订阅的例子（我用的是tomcat6.0，activeMq5.3.2）：

首先整合tomcat和activeMq：

1、准备jar包：

将ActiveMQ lib目录下的5个jar包复制到Tomcat lib目录下：

activemq-core-5.1.0.jar

activemq-web-5.1.0.jar

geronimo-j2ee-management_1.0_spec-1.0.jar

geronimo-jms_1.1_spec-1.1.1.jar

geronimo-jta_1.0.1B_spec-1.0.1.jar

2、修改配置文件：

2.1 修改Tomcat的conf/context.xml文件：

在<context></context>节点中添加以下内容：

<Resource

        name="jms/FailoverConnectionFactory"

        auth="Container"

        type="org.apache.activemq.ActiveMQConnectionFactory"

        description="JMS Connection Factory"

        factory="org.apache.activemq.jndi.JNDIReferenceFactory"

    brokerURL="failover:(tcp://localhost:61616)?initialReconnectDelay=100&amp;maxReconnectAttempts=5"

        brokerName="localhost"

        useEmbeddedBroker="false"/>

    <Resource

        name="jms/NormalConnectionFactory"

        auth="Container"

        type="org.apache.activemq.ActiveMQConnectionFactory"

        description="JMS Connection Factory"

        factory="org.apache.activemq.jndi.JNDIReferenceFactory"

        brokerURL="tcp://localhost:61616"

        brokerName="localhost"

        useEmbeddedBroker="false"/>

    <Resource name="jms/topic/MyTopic"

        auth="Container"

        type="org.apache.activemq.command.ActiveMQTopic"

        factory="org.apache.activemq.jndi.JNDIReferenceFactory"

        physicalName="MY.TEST.FOO"/>

    <Resource name="jms/queue/MyQueue"

        auth="Container"

        type="org.apache.activemq.command.ActiveMQQueue"

        factory="org.apache.activemq.jndi.JNDIReferenceFactory"

        physicalName="MY.TEST.FOO.QUEUE"/>

配置说明：以JNDI的方式定义了ActiveMQ的broker连接url、Topic和Queue。

此处需加以注意的是Listener端的borkerURL使用了failover传输方式：

failover:(tcp://localhost:61616)?initialReconnectDelay=100&amp;maxReconnectAttempts=5

客户端使用普通传输方式：tcp://localhost:61616

failover transport是一种重新连接机制，用于建立可靠的传输。此处配置的是一旦ActiveMQ broker中断，Listener端将每隔100ms自动尝试连接，直至成功连接或重试5次连接失败为止。

failover还支持多个borker同时提供服务，实现负载均衡的同时可增加系统容错性，格式： failover:(uri1,...,uriN)?transportOptions

2.2 在Myeclipse中新建一个jms-test的web工程

在src下，新建一个包com.flvcd.servlet （我想这个你应该会吧，如果不会，我只能说，你不合适干）

然后在修改web.xml，在里面加上：

<servlet>

<servlet-name>jms-listener</servlet-name>

<servlet-class>

com.flvcd.servlet.JMSListener

</servlet-class>

<load-on-startup>1</load-on-startup>

</servlet>

2.3 修改activemq.xml文件：为了支持持久化消息，需修改ActiveMQ的配置文件如下，使用默认的AMQ Message Store方式（索引文件方式）存储消息，据官网介绍是快速、稳定的。数据库存储方式可参照官网相关文档。

<broker xmlns="http://activemq.apache.org/schema/core" brokerName="localhost" persistent="true" useShutdownHook="false">

<persistenceAdapter>

<amqPersistenceAdapter directory="activemq-data" maxFileLength="32mb"/> </persistenceAdapter>

</broker>

(备注：2.3中红色部分直接加到后面就行，不用修改)

然后在src下的包里新建一个类，类名是JMSListener.java

再复制一下代码到你的类中：

package com.flvcd.servlet;



import java.io.*;

import javax.servlet.*;

import javax.servlet.http.*;

import javax.naming.*;

import javax.jms.*;

import org.apache.activemq.ActiveMQConnectionFactory;



public class JMSListener extends HttpServlet implements MessageListener {

   

    public void init(ServletConfig config) throws ServletException {

        try {

            InitialContext initCtx = new InitialContext();

            Context envContext = (Context) initCtx.lookup("java:comp/env");

            ConnectionFactory connectionFactory = (ConnectionFactory) envContext

                    .lookup("jms/FailoverConnectionFactory");

            Connection connection = connectionFactory.createConnection();

            connection.setClientID("MyClient");

            Session jmsSession = connection.createSession(false,

                    Session.AUTO_ACKNOWLEDGE);

            // 普通消息订阅者，无法接收持久消息 //MessageConsumer consumer =

            // jmsSession.createConsumer((Destination)

            // envContext.lookup("jms/topic/MyTopic"));

            // //基于Topic创建持久的消息订阅者，前提：Connection必须指定一个唯一的clientId，当前为MyClient

        TopicSubscriber consumer=jmsSession.createDurableSubscriber((Topic)envContext.lookup("jms/topic/MyTopic"), "MySub");

            consumer.setMessageListener(this);

            connection.start();

        } catch (NamingException e) {

            e.printStackTrace();

        } catch (JMSException e) {

            e.printStackTrace();

        }

    }



   

    public void onMessage(Message message) {

        if (checkText(message, "RefreshArticleId") != null) {

            String articleId = checkText(message, "RefreshArticleId");

            System.out.println("接收刷新文章消息，开始刷新文章ID=" + articleId);

        } else if (checkText(message, "RefreshThreadId") != null) {

            String threadId = checkText(message, "RefreshThreadId");

            System.out.println("接收刷新论坛帖子消息，开始刷新帖子ID=" + threadId);

        } else {

            System.out.println("接收普通消息，不做任何处理！");

        }

    }



    private static String checkText(Message m, String s) {

        try {

            return m.getStringProperty(s);

        } catch (JMSException e) {

            e.printStackTrace(System.out);

            return null;

        }

    }

}



你的监听端建好

现在开始建立你发布端

同样是新建一个类MyPublish.java

然后复制一下代码到你的类里

package com.flvcd.servlet;



import java.io.IOException;

import java.io.PrintWriter;



import javax.jms.Connection;

import javax.jms.ConnectionFactory;

import javax.jms.DeliveryMode;

import javax.jms.Destination;

import javax.jms.JMSException;

import javax.jms.Message;

import javax.jms.MessageListener;

import javax.jms.MessageProducer;

import javax.jms.Session;

import javax.naming.Context;

import javax.naming.InitialContext;

import javax.naming.NamingException;

import javax.servlet.ServletException;

import javax.servlet.http.HttpServlet;

import javax.servlet.http.HttpServletRequest;

import javax.servlet.http.HttpServletResponse;



public class MyPublish extends HttpServlet implements MessageListener {

   

   

    //定义初始化所需要的变量

    private InitialContext initCtx;

    private Context envContext;

    private ConnectionFactory connectionFactory;

    private Connection connection;

    private Session jmsSession;

    private MessageProducer producer;

   

   

    public void onMessage(Message message) {

        // TODO Auto-generated method stub



    }



   

    public MyPublish() {

        super();

    }



   

    public void destroy() {

        super.destroy(); // Just puts "destroy" string in log

        // Put your code here

    }



   

    public void doGet(HttpServletRequest request, HttpServletResponse response)

            throws ServletException, IOException {



    doPost(request, response);

    }



   

    public void doPost(HttpServletRequest request, HttpServletResponse response)

            throws ServletException, IOException {

        String content=request.getParameter("content");

        //设置持久方式

        try {

            producer.setDeliveryMode(DeliveryMode.PERSISTENT);

            Message testMessage = jmsSession.createMessage();

            // 发布刷新文章消息

            testMessage.setStringProperty("RefreshArticleId", content);

            producer.send(testMessage);

            // 发布刷新帖子消息

            testMessage.clearProperties();

            testMessage.setStringProperty("RefreshThreadId", content);

            producer.send(testMessage);

        } catch (Exception e) {

            e.printStackTrace();

        }

      

      

    }



   

    public void init() throws ServletException {

        // Put your code here

        try {

            initCtx = new InitialContext();

            envContext = (Context) initCtx.lookup("java:comp/env");

            connectionFactory = (ConnectionFactory) envContext.lookup("jms/NormalConnectionFactory");

            connection = connectionFactory.createConnection();

            jmsSession = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);

            producer = jmsSession.createProducer((Destination) envContext.lookup("jms/topic/MyTopic"));



        } catch (NamingException e) {

            e.printStackTrace();

        } catch (JMSException e) {

            e.printStackTrace();

        }

    }

}







最后，建一个MyPublish.jsp页面，加上一下代码

<form action="myPublish.do">

        <input type="text" name="content" />

        <input type="submit" value="提交" >

    </form>

在web.xml中加上一下代码

   

  <servlet>

    <description>This is the description of my J2EE component</description>

    <display-name>This is the display name of my J2EE component</display-name>

    <servlet-name>MyPublish</servlet-name>

    <servlet-class>com.flvcd.servlet.MyPublish</servlet-class>

  </servlet>

  <servlet>

  <servlet-mapping>

    <servlet-name>MyPublish</servlet-name>

    <url-pattern>/myPublish.do</url-pattern>

  </servlet-mapping>



将你的工程装载到你的tomcat中

然后先启动activeMq（前面已说明启动方式），然后用Myeclipse启动tomcat，在页面上输入http://localhost：8080/jms-test/MyPublish.jsp，在页面上输入一些英文字符，然后你在后台可以看到你输入的字符，

注意事项，在Myeclipse中导入activeMq的jar包，我用是5.3所以有一个activemq-all-5.3.2.jar，这个jar包是全部的内容都包括，直接用Myeclipse导包方法导入的。