【Java】ActiveMq5.8与Tomcat7相集成在MyEclipse10中的JMS测试

JMS:Java消息服务(Java Message Service),是sun提出来的为J2EE提供企业消息处理的一套规范,它提供了创建、发送、接收、读取消息的服务。能接受消息生产者(Message Provider)发出的消息,并把消息转发给消息消费者(MessageConsumer)对于JMS的文章,请参考其他资料

一、下载部署

1 . 安装

http://activemq.apache.org/download.html 下载5.8.0发行包,解压即可,

2 . 启动

window环境运行解压目录下的/bin/activemq.bat

3 . 测试

ActiveMQ默认使用的TCP连接端口是61616,通过查看该端口的信息可以测试ActiveMQ是否成功启动
window环境运行  netstat -an|find "61616"

4 . 监控

ActiveMQ5.8版本默认启动时,启动了内置的jetty服务器,提供一个demo应用和用于监控ActiveMQ的admin应用(消息管理后台系统)。admin:http://127.0.0.1:8161/admin/

二、Queue 与Topic 比较

1、JMS Queue执行load balancer语义:

一条消息仅能被一个consumer收到。如果在message发送的时候没有可用的consumer,那么它将被保存一直到能处理该message的consumer可用。如果一个consumer收到一条message后却不响应它,那么这条消息将被转到另一个consumer那儿。一个Queue可以有很多consumer,并且在多个可用的consumer中负载均衡。

2、Topic实现publish和subscribe语义:

一条消息被publish时,它将发到所有感兴趣的订阅者,所以零到多个subscriber将接收到消息的一个拷贝。但是在消息代理接收到消息时,只有激活订阅的subscriber能够获得消息的一个拷贝。

3、分别对应两种消息模式:

Point-to-Point (点对点),Publisher/Subscriber Model (发布/订阅者) , 其中在Publicher/Subscriber模式下又有Nondurablesubscription(非持久订阅)和durablesubscription (持久化订阅)2种消息处理方式。

三、Point-to-Point (点对点)消息模式开发流程

1、生产者(producer)开发流程(ProducerTool.java):

1.1 创建Connection

根据url,user和password创建一个jms Connection。

1.2 创建Session

在connection的基础上创建一个session,同时设置是否支持事务和ACKNOWLEDGE标识。

1.3 创建Destination对象

需指定其对应的主题(subject)名称,producer和consumer将根据subject来发送/接收对应的消息。

1.4 创建MessageProducer

根据Destination创建MessageProducer对象,同时设置其持久模式。

1.5 发送消息到队列(Queue)

封装TextMessage消息,使用MessageProducer的send方法将消息发送出去。

2、消费者(consumer)开发流程(ConsumerTool.java)

2.1 实现MessageListener接口

消费者类必须实现MessageListener接口,然后在onMessage()方法中监听消息的到达并处理。

2.2 创建Connection

根据url,user和password创建一个jms Connection,如果是durable模式,还需要给connection设置一个clientId。

2.3 创建Session和Destination

与ProducerTool.java中的流程类似,不再赘述。

2.4创建replyProducer【可选】

可以用来将消息处理结果发送给producer。

2.5 创建MessageConsumer

根据Destination创建MessageConsumer对象

四、ActiveMQ5.8与Tomcat7整合

网上有些整合可能需要其他包,但经本人测试只需要\apache-activemq-5.8.0\activemq-all-5.8.0.jar  一个包即可,在MyEclipse10中使用加载

修改Tomcat的conf/context.xml文件,在<context></context>节点中添加以下内容

<Resource suseEmbeddedBroker="false" brokerName="localhost" brokerURL="failover:(tcp://localhost:61616)?initialReconnectDelay=100&maxReconnectAttempts=5" factory="org.apache.activemq.jndi.JNDIReferenceFactory" description="JMS Connection Factory" type="org.apache.activemq.ActiveMQConnectionFactory" auth="Container" name="jms/FailoverConnectionFactory"/> 

<Resource brokerName="localhost" brokerURL="tcp://localhost:61616" factory="org.apache.activemq.jndi.JNDIReferenceFactory" description="JMS Connection Factory" type="org.apache.activemq.ActiveMQConnectionFactory" auth="Container" name="jms/NormalConnectionFactory" useEmbeddedBroker="false"/> 

<Resource factory="org.apache.activemq.jndi.JNDIReferenceFactory" type="org.apache.activemq.command.ActiveMQTopic" auth="Container" name="jms/topic/MyTopic" physicalName="MY.TEST.FOO"/> 

<Resource factory="org.apache.activemq.jndi.JNDIReferenceFactory" type="org.apache.activemq.command.ActiveMQQueue" auth="Container" name="jms/queue/MyQueue" 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

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

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

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

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

有些网站的朋友的代码修改是这样的,我试了一下有异常,貌似找不到activemq-data,具体的了解还不是很深

<broker xmlns="http://activemq.apache.org/schema/core" brokerName="localhost" persistent="true" useShutdownHook="false"> 
  <persistenceAdapter> 
    <amqPersistenceAdapter directory="activemq-data" maxFileLength="32mb"/>
  </persistenceAdapter>
</broker>

新建Web项目,目录结构如下

Listener端(JMSListener.java)完整实现

package com.servlet;  
  
import javax.servlet.*;  
import javax.servlet.http.*;  
import javax.naming.*;  
import javax.jms.*;  
  
@SuppressWarnings("serial")  
public class JMSListener extends HttpServlet implements MessageListener {  
    /** 初始化jms连接,创建topic监听器 */  
    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;  
        }  
    }  
}

创建一个Servlet类,MyPublish.java

package com.servlet;  
  
import java.io.IOException;  
  
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;  
  
@SuppressWarnings("serial")  
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) {  
    }  
  
    public void init() throws ServletException {  
        try {  
            initCtx = new InitialContext();  
            // 连接JNDI获得对象  
            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();  
        }  
    }  
      
    public void doGet(HttpServletRequest request, HttpServletResponse response)  
            throws ServletException, IOException {  
        doPost(request, response);  
    }  
  
    public void doPost(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {  
        String s = request.getParameter("content");  
        String content=new String(s.getBytes("iso8859-1"),"utf-8");  
        // 设置持久方式  
        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();  
        }  
    }  
}

最后,建一个MyPublish.jsp页面

<%@ page language="java" import="java.util.*" pageEncoding="utf-8"%>  
 <form action="myPublish.do">  
        <input type="text" name="content" />  
        <input type="submit" value="提交" >  
</form>

Web.xml文件配置如下

<?xml version="1.0" encoding="UTF-8"?>  
<web-app version="2.5" xmlns="http://java.sun.com/xml/ns/javaee"  
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"  
    xsi:schemaLocation="http://java.sun.com/xml/ns/javaee   http://java.sun.com/xml/ns/javaee/web-app_2_5.xsd">  
      
    <servlet>  
        <servlet-name>jms-listener</servlet-name>  
        <servlet-class>com.servlet.JMSListener</servlet-class>  
        <load-on-startup>1</load-on-startup>  
    </servlet>  
      
    <servlet>  
        <servlet-name>MyPublish</servlet-name>  
        <servlet-class>com.servlet.MyPublish</servlet-class>  
    </servlet>  
    <servlet-mapping>  
        <servlet-name>MyPublish</servlet-name>  
        <url-pattern>/myPublish.do</url-pattern>  
    </servlet-mapping>  
      
    <welcome-file-list>  
        <welcome-file>index.jsp</welcome-file>  
    </welcome-file-list>  
      
    <login-config>  
        <auth-method>BASIC</auth-method>  
    </login-config>  
</web-app>


然后先启动activeMq(前面已说明启动方式),然后用Myeclipse启动tomcat,在页面上输入http://localhost:8080/JMS-Test/MyPublish.jsp,在页面上输入一些英文字符,然后你在后台可以看到你输入的字符,

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


赞(52) 打赏
未经允许不得转载:优客志 » JAVA开发
分享到:

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

微信扫一扫打赏