ActiveMQ與Tomcat整合教程資料

說明：Tomcat示例版本6.0.14，其它版本在配置上可能有一些差異1、準(zhǔn)備jar包：將ActiveMQlib目錄下的5個(gè)jar包復(fù)制到Tomcatlib目錄下：activemq-core-5.1.0.jaractivemq-web-5.1.0.jargeronimo-j2ee-management_1.0_spec-1.0.jargeronimo-jms_1.1_spec-1.1.1.jargeronimo-jta_1.0.1B_spec-1.0.1.jar2、修改配置文件：2.1修改Tomcat的conf/context.xml文件：

在<context></context>節(jié)點(diǎn)中添加以下內(nèi)容：

<Resource

name="jms/FailoverConnectionFactory"

auth="Container"

type="org.apache.activemq.ActiveMQConnectionFactory"

description="JMSConnectionFactory"

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

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

brokerName="localhost"

useEmbeddedBroker="false"/>

<Resource

name="jms/NormalConnectionFactory"

auth="Container"

type="org.apache.activemq.ActiveMQConnectionFactory"

description="JMSConnectionFactory"

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

brokerURL="tcp://localhost:61616"

brokerName="localhost"

useEmbeddedBroker="false"/>

<Resourcename="jms/topic/MyTopic"

auth="Container"

type="mand.ActiveMQTopic"

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

physicalName="MY.TEST.FOO"/>

<Resourcename="jms/queue/MyQueue"

auth="Container"

type="mand.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&maxReconnectAttempts=5客戶端使用普通傳輸方式：tcp://localhost:61616failovertransport是一種重新連接機(jī)制，用于建立可靠的傳輸。此處配置的是一旦ActiveMQbroker中斷，Listener端將每隔100ms自動(dòng)嘗試連接，直至成功連接或重試5次連接失敗為止。failover還支持多個(gè)borker同時(shí)提供服務(wù)，實(shí)現(xiàn)負(fù)載均衡的同時(shí)可增加系統(tǒng)容錯(cuò)性，格式：

failover:(uri1,...,uriN)?transportOptions2.2新建web應(yīng)用(webapps/jms-test)，修改WEB-INF/web.xml文件：

增加一個(gè)自啟動(dòng)Servlet，該Servlet實(shí)現(xiàn)了MessageListener接口，作為Topic消息的Listener端。<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的配置文件如下，使用默認(rèn)的AMQMessageStore方式（索引文件方式）存儲消息，據(jù)官網(wǎng)介紹是快速、穩(wěn)定的。數(shù)據(jù)庫存儲方式可參照官網(wǎng)相關(guān)文檔。

<brokerxmlns="/schema/core"brokerName="localhost"persistent="true"useShutdownHook="false">

<persistenceAdapter>

<amqPersistenceAdapterdirectory="activemq-data"maxFileLength="32mb"/>

</persistenceAdapter>

</broker>3、Listener端（JMSListener.java）完整實(shí)現(xiàn)：packagecom.flvcd.servlet;importjava.io.*;

importjavax.servlet.*;

importjavax.servlet.http.*;

importjavax.naming.*;

importjavax.jms.*;

importorg.apache.activemq.ActiveMQConnectionFactory;publicclassJMSListenerextendsHttpServletimplementsMessageListener{

/**初始化jms連接，創(chuàng)建topic監(jiān)聽器*/

publicvoidinit(ServletConfigconfig)throwsServletException{

try{

InitialContextinitCtx=newInitialContext();

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

ConnectionFactoryconnectionFactory=(ConnectionFactory)envContext.lookup("jms/FailoverConnectionFactory");

Connectionconnection=connectionFactory.createConnection();

connection.setClientID("MyClient");

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

//普通消息訂閱者，無法接收持久消息

//MessageConsumerconsumer=jmsSession.createConsumer((Destination)envContext.lookup("jms/topic/MyTopic"));

//基于Topic創(chuàng)建持久的消息訂閱者，前提：Connection必須指定一個(gè)唯一的clientId，當(dāng)前為MyClient

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

consumer.setMessageListener(this);

connection.start();

}catch(NamingExceptione){

e.printStackTrace();

}catch(JMSExceptione){

e.printStackTrace();

}

}

/**接收消息，做對應(yīng)處理*/

publicvoidonMessage(Messagemessage){

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

StringarticleId=checkText(message,"RefreshArticleId");

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

}

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

StringthreadId=checkText(message,"RefreshThreadId");

System.out.println("接收刷新論壇帖子消息，開始刷新帖子ID="+threadId);

}else{

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

}

}

privatestaticStringcheckText(Messagem,Strings){

try{

returnm.getStringProperty(s);

}catch(JMSExceptione){

e.printStackTrace(System.out);

returnnull;

}

}

}編譯JMSListener.java至classes目錄：javac-cp.;D:\apache-tomcat-6.0.14\lib\servlet-api.jar;D:\apache-tomcat-6.0.14\lib\geronimo-jms_1.1_spec-1.1.1.jar;D:\apache-tomcat-6.0.14\lib\activemq-core-5.1.0.jar-d.JMSListener.java注：D:\apache-tomcat-6.0.14請?zhí)鎿Q成本地對應(yīng)目錄。4、Publisher端（publish.jsp）實(shí)現(xiàn)：在jms-test目錄下新建publish.jsp文件：

<%@pagelanguage="java"import="javax.jms.*"pageEncoding="GBK"%><%@pagelanguage="java"import="javax.naming.*"%><%@pagelanguage="java"import="org.apache.activemq.ActiveMQConnectionFactory"%>

<%

try{

InitialContextinitCtx=newInitialContext();

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

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

Connectionconnection=connectionFactory.createConnection();

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

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

//設(shè)置持久方式

producer.setDeliveryMode(DeliveryMode.PERSISTENT);

MessagetestMessage=jmsSession.createMessage();

//發(fā)布刷新文章消息

testMessage.setStringProperty("RefreshArticleId","2046");

producer.send(testMessage);

//發(fā)布刷新帖子消息

testMessage.clearProperties();

testMessage.setStringProperty("RefreshThreadId","331");

producer.send(testMessage);

}catch(NamingExceptione){

e.printStackTrace();

}catch(JMSExceptione){

e.printStackTrace();

}%>

Publisher和Listner之間通過Message的setStringProperty和getStringProperty方法，實(shí)現(xiàn)對應(yīng)的業(yè)務(wù)邏輯。上述示例代碼中，RefreshArticleId代表刷新某篇文章，RefreshThreadId代表刷新某個(gè)帖子，property值保持對應(yīng)的ID。當(dāng)然用戶可根據(jù)實(shí)際需求靈活地使用。5、運(yùn)行Demo：5.1啟動(dòng)ActiveMQ服務(wù)器5.2啟動(dòng)Tomcat服務(wù)器：JMSLi