首页 诗词 字典 板报 句子 名言 友答 励志 学校 网站地图
当前位置: 首页 > 教程频道 > 服务器 > 云计算 >

JMS ActiveMQ学习(1)

2012-09-14 
JMS ActiveMQ学习(一)这是本人第一次发表个人使用ActiveMq,如有错误,请指出。只为与众人共享我的心得。activ

JMS ActiveMQ学习(一)

      这是本人第一次发表个人使用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.6.0。

          2、安装

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

          3、启动ActiveMQ服务器

                直接运行bin\activemq.bat

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

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

       三、详细实例

            将tomcat和activeMq整合在一起做的一个发送接受的发布订阅的例子(我用的是tomcat6.0,activeMq5.6.0):

            首先整合tomcat和activeMq:

              1、准备jar包:

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

                    activemq-core-5.6.0.jar

                    activemq-web-5.6.0.jar

                    geronimo-j2ee-management_1.1_spec-1.0.1.jar

                    geronimo-jms_1.1_spec-1.1.1.jar

                    geronimo-jta_1.0.1B_spec-1.0.1.jar

               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>

           

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

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

package com.flvcd.servlet;

 

importjava.io.*;

import javax.servlet.*;

import javax.servlet.http.*;

import javax.naming.*;

import javax.jms.*;

importorg.apache.activemq.ActiveMQConnectionFactory;

 

publicclassJMSListener extends HttpServletimplements MessageListener {

    /**初始化jms连接,创建topic监听器*/

    publicvoid 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

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

            consumer.setMessageListener(this);

            connection.start();

        } catch (NamingException e) {

            e.printStackTrace();

        } catch (JMSException e) {

            e.printStackTrace();

        }

    }

 

    /**接收消息,做对应处理*/

    publicvoid onMessage(Message message) {

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

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

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

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

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

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

        } else {

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

        }

    }

 

    privatestatic String checkText(Message m, String s) {

        try {

            return m.getStringProperty(s);

        } catch (JMSException e) {

            e.printStackTrace(System.out);

            returnnull;

        }

    }

}

 

你的监听端建好

现在开始建立你发布端

同样是新建一个类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

 

    }

 

    /**

     * Constructor of the object.

     */

    public MyPublish() {

        super();

    }

 

    /**

     * Destruction of the servlet. <br>

     */

    public void destroy() {

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

        // Put your code here

    }

 

    /**

     * The doGet method of the servlet. <br>

     *

     * This method is called when a form has its tag value method equals to get.

     *

     * @param request the request send by the client to the server

     * @param response the response send by the server to the client

     * @throws ServletException if an error occurred

     * @throws IOException if an error occurred

     */

    public void doGet(HttpServletRequest request, HttpServletResponse response)

            throws ServletException, IOException {

 

    doPost(request, response);

    }

 

    /**

     * The doPost method of the servlet. <br>

     *

     * This method is called when a form has its tag value method equals to post.

     *

     * @param request the request send by the client to the server

     * @param response the response send by the server to the client

     * @throws ServletException if an error occurred

     * @throws IOException if an error occurred

     */

    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();

        }

       

       

    }

 

    /**

     * Initialization of the servlet. <br>

     *

     * @throws ServletException if an error occurs

     */

    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页面,加上一下代码

 <formaction="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.6所以有一个activemq-all-5.6.0.jar,这个jar包是全部的内容都包括,直接用Myeclipse导包方法导入的。

 

 

热点排行