程序喵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导包方法导入的
未经允许请勿转载：程序喵 » 【Java】ActiveMq5.8与Tomcat7相集成在MyEclipse10中的JMS测试