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&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导包方法导入的