[1]
在介紹ActiveMQ之前，首先簡要介紹一下JMS規範。
JMS的簡介:
(1)
JMS（Java Message Service，Java訊息服務）是一組Java應用程式介面(Java API)，它提供建立、傳送、接收、讀取訊息的服務。JMS 使您能夠通過訊息收發服務從一個 JMS 客戶機向另一個 JML 客戶機交流訊息。

JMS是一種與廠商無關的 API，用來訪問訊息收發系統。它類似於 JDBC (Java Database Connectivity)：這裡，JDBC 是可以用來訪問許多不同關係資料庫的 API，而 JMS 則提供同樣與廠商無關的訪問方法，以訪問訊息收發服務。許多廠商目前都支援 JMS，包括 IBM 的 MQSeries、BEA的 Weblogic JMS service和 Progress 的 SonicMQ.

(2)
JMS典型的應用場景:
操作可非同步執行.
發email了, 發msn訊息了.
或者一些比較耗時的操作, 比如要在某目錄下生成一個大報表. 操作者把指令發出去就完事.

[2]
JMS的基本構件:
(1)
Broker
什麼是Broker呢？可以把JMS Brokers 看成是伺服器端。這個伺服器可以獨立執行.也可以隨著其他容器
以內嵌方式雲心，如下配置：
使用顯示的Java程式碼建立
BrokerService broker = new BrokerService();
// configure the broker
broker.addConnector("tcp://localhost:61616");
broker.start();
使用BrokerFacotry建立
BrokerService broker = BrokerFactory.getInstance().createBroker(someURI);
使用Spring Bean建立
<bean id=”broker” class=”org.apache.activemq.xbean.BrokerFactoryBean”>
<property name=”config” value=”classpath:org/apache/activemq/xbean/activemq.xml” />
<property name=”start” value=”true” />
</bean>
還可以使用XBean或Spring 2.0等多種配置方式配置，
通過ActiveMQConnectionFactory還可以隱含的建立內嵌的broker，這個broker就不是一個獨立的服務了。
<bean id=”jmsTemplate” class=”org.springframework.jms.core.JmsTemplate”>
<property name=”connectionFactory” ref=”jmsFactory”/>
<property name=”defaultDestination” ref=”destination” />
<property name=”destinationResolver” ref=”預設是DynamicDestionResolver” />
<property name=”pubSubDomain”><value>true or false預設是false，
false是QueneDestination, true是TopicDestination</value>
</bean>
上面的defaultDestination是指預設傳送和接收的目的地，我們也可以不指定，而是通過目的地名稱讓jmsTemplate自動幫我們建立.

(2)
1 連線工廠
連線工廠是客戶用來建立連線的物件，例如ActiveMQ提供的ActiveMQConnectionFactory。
2 連線
JMS Connection封裝了客戶與JMS提供者之間的一個虛擬的連線。
3 會話
JMS Session是生產和消費訊息的一個單執行緒上下文。會話用於建立訊息生產者（producer）、訊息消費者（consumer）和訊息（message）等。會話提供了一個事務性的上下文，在這個上下文中，一組傳送和接收被組合到了一個原子操作中。
(3)
目的地:
目的地是客戶用來指定它生產的訊息的目標和它消費的訊息的來源的物件。JMS1.0.2規範中定義了兩種
訊息傳遞域：Point-to-Point訊息(P2P),點對點;釋出訂閱訊息(Publish Subscribe messaging，簡稱Pub/Sub)
兩者的區別:
P2P訊息模型是在點對點之間傳遞訊息時使用。如果應用程式開發者希望每一條訊息都能夠被處理，那麼應該使用P2P訊息模型。與Pub/Sub訊息模型不同，P2P訊息總是能夠被傳送到指定的位置。
P2P訊息,每個訊息只能有一個消費者。
　　Pub/Sub模型在一到多的訊息廣播時使用。如果一定程度的訊息傳遞的不可靠性可以被接受的話，那麼應用程式開發者也可以使用Pub/Sub訊息模型。換句話說，它適用於所有的訊息消費程式並不要求能夠收到所有的資訊或者訊息消費程式並不想接收到任何訊息的情況。
Pub/Sub,每個訊息可以有多個消費者。
在點對點訊息傳遞域中，目的地被成為佇列（queue）；在釋出/訂閱訊息傳遞域中，目的地被成為主題（topic）。
(3)
3.1
訊息生產者
訊息生產者是由會話建立的一個物件，用於把訊息傳送到一個目的地。
3.2
訊息消費者
訊息消費者是由會話建立的一個物件，它用於接收發送到目的地的訊息。訊息的消費可以採用以下兩種
方法之一：
? 非同步消費。客戶可以為消費者註冊一個訊息監聽器，以定義在訊息到達時所採取的動作。(非同步操作)
? 同步消費。通過呼叫消費者的receive方法從目的地中顯式提取訊息。receive方法可以一直阻塞到訊息到達。
3.3
訊息是 JMS 中的一種型別物件，由兩部分組成：報頭和訊息主體。報頭由路由資訊以及有關該訊息的元資料組成。訊息主體則攜帶著應用程式的資料或有效負載。根據有效負載的型別來劃分，可以將訊息分為幾種型別，它們分別攜帶：
簡單文字 (TextMessage)、可序列化的物件 (ObjectMessage)、屬性集合
(MapMessage)、位元組流 (BytesMessage)、原始值流 (StreamMessage)，還有無有效負載的訊息 (Message)。
(4)
JMS定義了從0到9的優先順序路線級別，0是最低的優先順序而9則是最高的。更特殊的是0到4是正常優先順序的變化幅度，而5到9是加快的優先順序的變化幅度。

[3]
ActiveMQ簡介:
ActiveMQ 是開源的JMS實現，Geronimo應用伺服器就是使用的ActiveMQ提供JMS服務。
安裝
在http://activemq.apache.org/download.html 下載5.0.0發行包，解壓即可，
啟動
window環境執行解壓目錄下的/bin/activemq.bat
測試
ActiveMQ預設使用的TCP連線埠是61616, 通過檢視該埠的資訊可以測試ActiveMQ是否成功啟動
window環境執行 netstat -an|find "61616"
監控
ActiveMQ5.0版本預設啟動時，啟動了內建的jetty伺服器，提供一個demo應用和用於監控ActiveMQ的admin應用。
admin：http://127.0.0.1:8161/admin/
demo：http://127.0.0.1:8161/demo/
點選demo應用中的“ Market data publisher ”，就會發一些測試的訊息。轉到admin頁面的topics menu下面（queue和topic的區別見 http://andyao.javaeye.com/blog/153173 ），可以看到訊息在增長。
ActiveMQ5.0的配置檔案在解壓目錄下的/conf目錄下面。主要配置檔案為activemq.xml.

[4]
例項一:(沒有結合spring框架)
public class QueueProducer {
/*
* 建立的簡圖
ConnectionFactory---->Connection--->Session--->Message
Destination + Session------------------------------------>Producer
Destination + Session------------------------------------>MessageConsumer
*/
public static void main(String[] args) {
// ConnectionFactory ：連線工廠，JMS 用它建立連線
ConnectionFactory connectionFactory;
// Connection ：JMS 客戶端到JMS Provider 的連線
Connection connection = null;
// Session： 一個傳送或接收訊息的執行緒
Session session;
// Destination ：訊息的目的地;訊息傳送給誰.
Queue queue;
//設定回覆的目的地
Queue replyQueue;
// MessageProducer：訊息傳送者
MessageProducer producer;
MessageConsumer replyer;
connectionFactory = new ActiveMQConnectionFactory(
ActiveMQConnection.DEFAULT_USER,
ActiveMQConnection.DEFAULT_PASSWORD,
"tcp://192.168.1.191:61616");
try {
// 構造從工廠得到連線物件
connection = connectionFactory.createConnection();
// 啟動
connection.start();
// 獲取操作連線
session = connection.createSession(Boolean.TRUE,Session.AUTO_ACKNOWLEDGE);
// 建立佇列: 可以在http://localhost:8161/admin/queue.jsp中看到
queue=new ActiveMQQueue("jason.queue2");
replyQueue=new ActiveMQQueue("jason.replyQueue");
// 得到訊息生成者【傳送者】:需要由Session和Destination來建立
producer = session.createProducer(queue);
// 建立訊息
TextMessage message = session.createTextMessage("jason學習ActiveMq 傳送的訊息");
//在訊息中設定回覆的目的地,
//對方用MessageProducer sender=session.createProducer(message.getJMSReplyTo());建立回覆者
message.setJMSReplyTo(replyQueue);
// 傳送一個non-Persistent的訊息
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
producer.send(message);
// 傳送一個Persistent的訊息
producer.setDeliveryMode(DeliveryMode.PERSISTENT);
producer.send(session.createTextMessage("這是一個Persistent的訊息!重啟JMS,仍可獲取"));
System.out.println("傳送訊息：jason學習ActiveMq 傳送的訊息");
System.out.println("這是一個Persistent的訊息!重啟JMS,仍可獲取");

//用回覆的目的地定義回覆接收者,且設定偵聽
replyer=session.createConsumer(replyQueue);
replyer.setMessageListener
(
new MessageListener()
{
public void onMessage(Message message)
{
try {
TextMessage txtmess = (TextMessage) message;
System.out.println("consumer的回覆內容是: "+txtmess.getText());
} catch (Exception e) {
e.printStackTrace();
}
}
}
);
session.commit();
} catch (Exception e) {
e.printStackTrace();
} finally {
try {
if (null != connection)
connection.close();
} catch (Throwable ignore) {
}
}
}
}
接收者:
//public class Receiver {
public class QueueConsumer implements MessageListener{
public static void main(String[] args)
{
QueueConsumer re=new QueueConsumer();
//迴圈只是為了讓程式每2秒進行一次連線偵聽是否有訊息可以獲取.
while(true)
{
re.consumeMessage();
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
//對於主動接收的,只須直接執行:re.consumeMessage();即可.
//其中的while(true),會一次性將所有的訊息獲取過來.
}

public void consumeMessage()
{
ConnectionFactory connectionFactory;
Connection connection = null;
Session session;
Queue queue;
MessageConsumer consumer;

connectionFactory = new ActiveMQConnectionFactory(
ActiveMQConnection.DEFAULT_USER,
ActiveMQConnection.DEFAULT_PASSWORD,
"tcp://192.168.1.191:61616");
try {
connection = connectionFactory.createConnection();
connection.start();
session = connection.createSession(Boolean.FALSE,
Session.AUTO_ACKNOWLEDGE);
queue=new ActiveMQQueue("jason.queue2");
consumer = session.createConsumer(queue);

// 接受訊息方式一:主動的去接受訊息,用consumer.receive
//只能獲取一條訊息 -->不採用
// TextMessage message = (TextMessage) consumer.receive(1000);
// if (null != message) {
// System.out.println("收到訊息" + message.getText());
// }
//可以不斷迴圈,獲取所有的訊息.--->關鍵.
// while (true) {
// TextMessage message = (TextMessage) consumer.receive(1000);
// if (null != message) {
// System.out.println("收到訊息" + message.getText());
// } else {
// break; //沒有訊息時,退出
// }
// }

/*接受訊息方式二:基於訊息監聽的機制，需要實現MessageListener介面，這個介面有個onMessage方法，當接受到訊息的時候會自動呼叫這個函式對訊息進行處理。
*/
consumer.setMessageListener(this);


} catch (Exception e) {
e.printStackTrace();
} finally {
try {
if (null != connection)
connection.close();
} catch (Throwable ignore) {
}
}
}

public void onMessage(Message message)
{
try {
if (message instanceof TextMessage) {
TextMessage txtmess = (TextMessage) message;
System.out.println("收到的訊息是:" + txtmess.getText());
//回覆傳送者
MessageProducer sender=session.createProducer(message.getJMSReplyTo());
sender.send(session.createTextMessage("已收到你的訊息"));
}
else
System.out.println("收到的訊息是:" + message);
} catch (Exception e) {
}
}
}
說明:
(2)
VM Transport
VM transport允許在VM內部通訊，從而避免了網路傳輸的開銷。這時候採用的連線不是socket連線，而是直接地方法呼叫。第一個建立VM 連線的客戶會啟動一個embed VM broker，接下來所有使用相同的broker name的VM連線都會使用這個broker。當這個broker上所有的連線都關閉的時候，這個broker也會自動關閉。
TCP Transport
TCP transport 允許客戶端通過TCP socket連線到遠端的broker。以下是配置語法：
tcp://hostname:port?transportOptions
tcp://localhost:61616
(3)
3.1
啟動activeMQ後,使用者建立的queues會被儲存在activeMQ解壓目錄下的\data\kr-store\data中.
3.2
建立queue,可以在程式碼中建立,也可以直接進入http://localhost:8161/admin--->點選queue--->在上面的field中填下你要建立的queue名-->點選建立即可.
3.3
若使用者建立的queue,不是持久化的,則在重啟activeMQ後,資料檔案中的內容會被清空,但檔案仍存在.
(4)
messageProducer傳送訊息後,會在儲存在目的地,即上面的queue中,也即就是在\data\kr-store\data目錄下的檔案中.
messageReceiver(連線到相同目的地的接收者),不需要立即接收.只要activeMQ的服務端不關閉,當執行接收者,連線到activeMQ的服務端時,就可以獲取activeMQ服務端上已傳送的訊息.
傳送/接收的訊息情況及數量及訊息的內容與處理(刪除),可以在
http://localhost:8161/admin/queue.jsp中檢視,操作.
(5)
可以將activeMQ的服務端放於一PC中,傳送者位於另一PC,接收者也位於另一PC中.
只要:tcp://activeMQ的服務端IP:activeMQ的服務埠,進行連線即可.
(6)
queue訊息,只被消費一次.

topic的例項(無結合spring)
public class TopicTest {
public static void main(String[] args) throws Exception {
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("vm://localhost");
Connection connection = factory.createConnection();
connection.start();
//建立一個Topic
Topic topic= new ActiveMQTopic("testTopic");
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//註冊消費者1
MessageConsumer comsumer1 = session.createConsumer(topic);
comsumer1.setMessageListener(new MessageListener(){
public void onMessage(Message m) {
try {
System.out.println("Consumer1 get " + ((TextMessage)m).getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
});
//註冊消費者2
MessageConsumer comsumer2 = session.createConsumer(topic);
comsumer2.setMessageListener(new MessageListener(){
public void onMessage(Message m) {
try {
System.out.println("Consumer2 get " + ((TextMessage)m).getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
});
//建立一個生產者，然後傳送多個訊息。
MessageProducer producer = session.createProducer(topic);
for(int i=0; i<10; i++){
producer.send(session.createTextMessage("Message:" + i));
}
}
}
輸出如下:
Consumer1 get Message:0
Consumer2 get Message:0
Consumer1 get Message:1
Consumer2 get Message:1
Consumer1 get Message:2
Consumer2 get Message:2
.............................

例項二:
非同步的電子郵件(將topic的應用例項也放在一起)
(4.1)
背景說明:　　
以傳統的方式傳送電子郵件（作為同步請求的一部分）會引起一些問題。首先，連線到電子郵件伺服器需要一次網路往返，速度可能會很慢，尤其是在伺服器非常繁忙的時候。過載的電子郵件伺服器甚至可以使依賴於電子郵件的服務暫時不可用。
xa 事務支援
另一個顯而易見的問題是，電子郵件伺服器通常在本質上是非事務性的。當事務被回滾時，這可以導致出現不一致的通知——把一條訊息放入佇列之後不能取消它。幸運的是， jms 支援事務，而且可以通過把訊息的傳送延遲到提交底層事務的時候來解決這個問題。
(4.2)
實現效果:
使用者在更改密碼後,系統會發送郵件通知使用者，為了避免傳送郵件時程式對使用者操作的阻塞，可以用JMS非同步傳送郵件.
(4.3)
實現流程:
當用戶更改密碼後
--1->呼叫JMS的傳送者,傳送者會利用jmsTemplate傳送訊息到目的地
--2->系統,執行原系統的程式.
--2->當訊息傳送後,messageListener偵聽到訊息,接收後執行相應的方法handleMessage().在此方法中執行傳送email.
好處:
從而實現了非同步傳送郵件,避免使用者等待單一執行緒完成，使原程式的執行更快,提升使用者體驗
<1>
第一步:
建立broker,jmsFactory,destination,messageConverter,jmsTemplate(用於傳送JMS訊息).
然後將jmsFactory,Destination,MessageConverter放入jmsTemplate中.
(1)
首先，我們在Spring中加入ActiveMQ Broker的配置：
1.
刪除原來jason-servlet.xml中的
<!DOCTYPE beans PUBLIC "-//SPRING//DTD BEAN 2.0//EN"
"http://www.springframework.org/dtd/spring-beans-2.0.dtd">
<beans>
改為:
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:amq="http://activemq.org/config/1.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-2.0.xsd
http://activemq.org/config/1.0
http://activemq.apache.org/schema/core/activemq-core-5.0.0.xsd">
同時,要匯入相應的jar包即可.
2.
<!-- 在Spring中配置嵌入式的 activemq broker,這樣就不用在使用JMS時,要手動啟動activeMQ broker -->
<amq:broker useJmx="false" persistent="false">
<amq:transportConnectors>
<amq:transportConnector uri="tcp://localhost:61616" />
</amq:transportConnectors>
</amq:broker>
<!-- 訊息的儲存機制, 伺服器重啟也不會丟失訊息.
<amq:broker useJmx="false" persistent="true">
<amq:persistenceAdapter>
<amq:amqPersistenceAdapter directory="d:/amq"/>
</amq:persistenceAdapter>
<amq:transportConnectors>
<amq:transportConnector uri="tcp://localhost:61616" />
<amq:transportConnector uri="vm://localhost:0" />
</amq:transportConnectors>
</amq:broker>
-->
(2)
在Spring中配置JMS Connection Factory。
<bean id="jmsFactory2"
class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="tcp://localhost:61616" />
</bean>
也可直接用:
<amq:connectionFactory id="jmsFactory2" brokerURL="tcp://localhost:61616" />
注意其中的borkerURL，應該是你在activemq.xml中transportconnector節點的uri屬性，這表示JMS Server的監聽地址。activeMQ預設埠是61616,由於採用預設方式,所以這裡也是61616.
同時,要執行程式之前,我們要先啟動broker,即啟動解壓目錄下的/bin/activemq.bat.
(3)
配置訊息傳送目的地：
<bean id="topicDestination"
class="org.apache.activemq.command.ActiveMQTopic">
<constructor-arg value="MY.topic" />
</bean>
<bean id="queueDestination"
class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg value="MY.queue" />
</bean>
也可直接用:
<amq:topic name="topicDestination" physicalName="MY.topic"/>
<amq:queue name="queueDestination" physicalName="MY.queue"/>
在 JMS中，目的地有兩種：主題(topic)和佇列(queue)。兩者的區別是：當一個主題目的地中被放入了一個訊息後，所有的訂閱者都會收到通知；而 對於佇列，僅有一個“訂閱者”會收到這個訊息，佇列中的訊息一旦被處理，就不會存在於佇列中。顯然，對於郵件傳送程式來說，使用佇列才是正確的選擇，而使 用主題時，可能會發送多封相同的郵件。
(4)
配置Spring中訊息傳送的JMS Template:
(與hibernate相似,其配置的是hibernateTemplate.都要將連線工廠放到template中)
<bean id="producerJmsTemplate"
class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory">
<bean class="org.springframework.jms.connection.SingleConnectionFactory">
<property name="targetConnectionFactory"
ref="jmsFactory2" />
</bean>
</property>
<!-- 因為要實現此模板同時用於queue與topic,所以目的地要放於傳送者中.若單獨使用,放於此,更方便!
<property name="defaultDestination" ref="queueDestination" />
-->
<property name="messageConverter" ref="userMessageConverter" />
</bean>
(5)
在實際的訊息傳送中，郵件內容需要用到User.username, User.password, User.email, User.fullname，我們定義了messageConverter,在傳送資訊時,將user物件轉換成訊息,在接收訊息時,會將訊息轉換成 User物件.只要在上面的jmstemplate中設定了messageConverter屬性,傳送/接收訊息時,Spring會自動幫我們進行轉 換，下面是Converter的配置和程式碼：
<bean id="userMessageConverter"
class="com.laoer.bbscs.jms.UserMessageConverter" />
程式碼如下:
public class UserMessageConverter implements MessageConverter {
private static transient Log logger = LogFactory.getLog(UserMessageConverter.class);
public Object fromMessage(Message message) throws JMSException {
if (logger.isDebugEnabled()) {
logger.debug("Receive JMS message: " + message);
}
if (message instanceof ObjectMessage) {
ObjectMessage oMsg = (ObjectMessage) message;
if (oMsg instanceof ActiveMQObjectMessage) {
ActiveMQObjectMessage aMsg = (ActiveMQObjectMessage) oMsg;
try {
PersonInfo personInfo = (PersonInfo) aMsg.getObject();
return personInfo;
} catch (Exception e) {
logger.error("Message:[" + message + "] is not a instance of personInfo.");
throw new JMSException("Message:[" + message + "] is not a instance of personInfo.");
}
} else {
logger.error("Message:[" + message + "] is not " + "a instance of ActiveMQObjectMessage[personInfo].");
throw new JMSException("Message:[" + message + "] is not " + "a instance of ActiveMQObjectMessage[personInfo].");
}
} else {
logger.error("Message:[" + message + "] is not a instance of ObjectMessage.");
throw new JMSException("Message:[" + message + "] is not a instance of ObjectMessage.");
}
}

public Message toMessage(Object obj, Session session) throws JMSException {
if (logger.isDebugEnabled()) {
logger.debug("Convert User object to JMS message: " + obj);
}
if (obj instanceof PersonInfo) {
ActiveMQObjectMessage msg = (ActiveMQObjectMessage) session.createObjectMessage();
msg.setObject((PersonInfo) obj);
return msg;
} else {
logger.error("Object:[" + obj + "] is not a instance of PersonInfo.");
throw new JMSException("Object:[" + obj + "] is not a instance of PersonInfo.");
}
}
}
此程式實現了MessageConverter介面，並實現其中的fromMessage和toMessage方法，分別實現轉換接收到的訊息為User物件和轉換User物件到訊息。
我們在程式中使用的是ActiveMQObjectMessage，它是ActiveMQ中對javax.jms.ObjectMessage的一個實現。
<2>
第二步:
配置傳送訊息:
(1)
在spring配置檔案中配置傳送者:
<!-- 傳送queue訊息 -->
<bean id="userMessageProducer" class="com.laoer.bbscs.jms.UserMessageProducer">
<property name="jmsTemplate" ref="producerJmsTemplate" />
<property name="defaultDestination" ref="queueDestination" />
</bean>
<!-- 傳送topic訊息 -->
<bean id="topicMessageProducer" class="com.laoer.bbscs.jms.TopicMessageProducer">
<property name="jmsTemplate" ref="producerJmsTemplate" />
<property name="defaultDestination" ref="topicDestination" />
</bean>
由於傳送者是用jmsTemplate進行傳送的,所以有注入jmsTemplate.
(2)
程式碼如下:
package com.laoer.bbscs.jms;
public class UserMessageProducer {
private JmsTemplate jmsTemplate;
//因為"目的地"屬性放於jmsTemplate中,則不用新增此屬性.
private Queue defaultDestination;
public void sendUserLoginInformationMail(PersonInfo personInfo) {
// 若"目的地"屬性放於jmsTemplate中,則用此方式
// getJmsTemplate().convertAndSend(personInfo);
getJmsTemplate().convertAndSend(this.defaultDestination,personInfo);
}
getter,setter略...
}
其中,sendUserLoginInformationMail方法是唯一我們需要編寫的，呼叫JMSTemplate的convertAndSend方法，Spring會自己呼叫我們之前配置的converter來轉換我們傳送的User物件,再進行傳送。
public class TopicMessageProducer {
private JmsTemplate jmsTemplate;
private Topic defaultDestination;
public void sendTopicMessage(PersonInfo personInfo) {
getJmsTemplate().convertAndSend(this.defaultDestination,personInfo);
}
getter,setter略...
}

<3>
實現訊息的接收
我們使用MDP（Message Drive POJO）來實現訊息的非同步接收。
我們需要實現javax.jms.MessageListener介面的void onMessage(Message message)方法來接收訊息。
不過我們可以使用Spring中提供的MessageListenerAdapter來簡化接收訊息的程式碼。
(0)
<!-- 定義訊息消費者,然後直接在messageListener中呼叫.
消費者,不用加入jmsTemplate屬性,jmsTemplate只用於傳送訊息 -->
<!-- queue訊息消費者,只能一個 -->
<bean id="userMessageConsumer" class="com.laoer.bbscs.jms.UserMessageConsumer">
<property name="mailSender" ref="mailSender" />
</bean>
<!-- topic訊息消費者,可以多個 -->
<bean id="topicConsumerA" class="com.laoer.bbscs.jms.TopicConsumerA" />
<bean id="topicConsumerB" class="com.laoer.bbscs.jms.TopicConsumerB" />
(1)
配置messageListener,用它來接收訊息.(用jmsTemplate來發送訊息)
<!-- 定義queue,topic(A/B consumer)各自的偵聽器 -->
<bean id="messageListener"
class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
<constructor-arg ref="userMessageConsumer"/>
<property name="defaultListenerMethod" value="handleMessage" />
<property name="messageConverter" ref="userMessageConverter" />
</bean>
<bean id="topicListenerA"
class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
<constructor-arg ref="topicConsumerA"/>
<property name="defaultListenerMethod" value="receiveA" />
<property name="messageConverter" ref="userMessageConverter" />
</bean>
<bean id="topicListenerB"
class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
<constructor-arg ref="topicConsumerB"/>
<property name="defaultListenerMethod" value="receiveB" />
<property name="messageConverter" ref="userMessageConverter" />
</bean>
其中的mailSender:即是我們的郵件傳送類，此類中的方法send()方法實現了郵件傳送的功能。
訊息偵聽介面卡defaultListenerMethod屬性:指定Spring在收到訊息後呼叫的方法，此處為handleMessage，Spring會根據收到的訊息--轉換為User物件-->呼叫handleMessage()方法。
(2)
配置訊息偵聽容器，並指定我們定義的訊息偵聽器。
<!-- 定義queue,topic(A/B consumer)各自的偵聽容器 -->
<bean id="listenerContainer"
class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="concurrentConsumers" value="5" />
<property name="connectionFactory" ref="jmsFactory2" />
<property name="destination" ref="queueDestination" />
<property name="messageListener" ref="messageListener" />
</bean>
<bean id="topicListenerContainerA"
class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="concurrentConsumers" value="5" />
<property name="connectionFactory" ref="jmsFactory2" />
<property name="destination" ref="topicDestination" />
<property name="messageListener" ref="topicListenerA" />
</bean>
<bean id="topicListenerContainerB"
class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="concurrentConsumers" value="5" />
<property name="connectionFactory" ref="jmsFactory2" />
<property name="destination" ref="topicDestination" />
<property name="messageListener" ref="topicListenerB" />
</bean>
配置了同時偵聽的個數,連線工廠(傳送者,接收者都要連線到同一地方),目的地(與上面對應),自定義的偵聽器.
(3)
接收者的程式碼:
package com.laoer.bbscs.jms;
public class UserMessageConsumer {
private static transient Log logger = LogFactory.getLog(UserMessageConsumer.class);
private MailSender mailSender;
public void handleMessage(PersonInfo personInfo) throws JMSException {
if (logger.isDebugEnabled()) {
logger.debug("Receive a User object from ActiveMQ: " + personInfo.toString());
}
mailSender.send(personInfo, "h***

[email protected]");
}
getter,setter略....
}

public class TopicConsumerA {
public void receiveA(PersonInfo personInfo) throws JMSException
{
System.out.println("TopicConsumerA收到TopicProducer的訊息---->personInfo的使用者名稱是:"+personInfo.getName());
}
}
TopicConsumerB與TopicConsumerA相似.

(4)
傳送郵件的相應方法:
public String send(PersonInfo personinfo,String mailAddr)
{
System.out.println("現在的時間是: "+System.currentTimeMillis());
// 不要使用SimpleEmail,會出現亂碼問題
HtmlEmail email = new HtmlEmail();
try{
// 這裡是傳送伺服器的名字
email.setHostName("smtp.sohu.com");
// 編碼集的設定
email.setCharset("gbk");
// 收件人的郵箱
// email.addTo("h***[email protected]");
email.addTo(mailAddr);
// 傳送人的郵箱
email.setFrom("sh****[email protected]", "she**fa");
// 如果需要認證資訊的話，設定認證：使用者名稱-密碼。分別為發件人在郵件
伺服器上的註冊名稱和密碼
email.setAuthentication("she**fa","20***23");
email.setSubject("測試email與JMS--你的密碼修改了");
// 要傳送的資訊
email.setMsg("你現在的使用者名稱是:"+personinfo.getName()+" \n密碼
是:"+personinfo.getPassword());
// 傳送
email.send();
System.out.println("現在的時間是: "+System.currentTimeMillis());
return "傳送成功!";
} catch (EmailException e) {
return "傳送失敗!";
}
}
(5)
在jLogin.java中呼叫即可(也新增相應的producer及setter,getter):
//用ActiveMQ的queue,正常發email
this.getMailSender().send(pi, hw***@126.com");
this.getUserMessageProducer().sendUserLoginInformationMail(pi);
//用ActiveMQ的topic釋出訊息
this.getTopicMessageProducer().sendTopicMessage(pi);

這樣,當執行後,會非同步傳送郵件,同時topicProducer傳送一個訊息後,topicConsumerA/B都會偵聽接收到,且執行相應的操作.
[6]
例項二:
整合ActiveMQ,Quartz,實現定時傳送郵件:
(1)
activeMQ的配置同上;
(2)
如何將ActiveMQ整合到Quartz中去:
1.用非繼承的方法實現quartz有任務類.
將任務類的目標類,目標方法:指定為訊息傳送類,及其傳送訊息的方法即可.
<!-- 利用Quartz定時分email -->
<bean id="mailJobDetailBean"
class="org.springframework.scheduling.quartz.MethodInvokingJobDetailFactoryBean">
<property name="targetObject">
<ref local="userMessageProducer"/> -->指定傳送者
</property>
<property name="targetMethod"> -->指定傳送方法,不同於上面
<value>send2</value>
</property>
</bean>
public void send2()
{
PersonInfo personInfo=new PersonInfo();
personInfo.setName("jason");
personInfo.setPassword("123456789");
this.getJmsTemplate().convertAndSend(this.defaultDestination,personInfo);
}
2.建立時間表.
<bean d="mailCronTriggerBean"
class="org.springframework.scheduling.quartz.CronTriggerBean">
<property name="jobDetail">
<ref local="mailJobDetailBean"/>
</property>
<property name="cronExpression">
<value>30 * * * * ?</value>
</property>
</bean>
3.放入任務工廠.
<bean id="schedulerFactoryBean"
class="org.springframework.scheduling.quartz.SchedulerFactoryBean">
<property name="triggers">
<list>
<ref bean="cronTriggerBean"/>
<ref bean="jasonGetPersonCronTriggerBean"/>
<ref bean="mailCronTriggerBean"/>
</list>
</property>
</bean>


1 DestinationResolver
DestinationResolver介面的作用是將指定的目的地名解析為目的地例項。其定義如下：
Java程式碼
1. public interface DestinationResolver {
2. Destination resolveDestinationName(Session session, String destinationName,
3. boolean pubSubDomain) throws JMSException;
4. }
引數pubSubDomain用於指定是使用“釋出/訂閱”模式（解析後的目的地是Topic），還是使用“點對點”模式（解析後的目的地是Queue）。

CachingDestinationResolver介面繼承了DestinationResolver，增加了快取的功能，其介面定義如下：
Java程式碼
1. public interface CachingDestinationResolver extends DestinationResolver {
2. void removeFromCache(String destinationName);
3. void clearCache();
4. }
在目的地失效的時候，removeFromCache方法會被呼叫；在JMS provider失效的時候，clearCache方法會被呼叫。

1.1 DynamicDestinationResolver
DynamicDestinationResolver實現了DestinationResolver介面。根據指定的目的地名，DynamicDestinationResolver會動態建立目的地例項。針對JMS1.1規範，它採用如下方法建立目的地：
Java程式碼
1. session.createTopic(topicName)
2. session.createQueue(queueName);

1.2 JndiDestinationResolver
JndiDestinationResolver繼承自JndiLocatorSupport， 同時實現了CachingDestinationResolver介面。如果在JMS provider中配置了靜態目的地，那麼JndiDestinationResolver通過JNDI查詢的方式獲得目的地例項。

JndiDestinationResolver的fallbackToDynamicDestination屬性用於指定在JNDI查詢失敗後，是否使 用動態目的地，預設值是false。JndiDestinationResolver的cache屬性用於指定是否對目的地例項進行快取，預設值是 true。

1.3 BeanFactoryDestinationResolver
BeanFactoryDestinationResolver實現了DestinationResolver介面和BeanFactoryAware介面。它會根據指定的目的地名從BeanFactory中查詢目的地例項。以下是相關的程式碼：
Java程式碼
1. public Destination resolveDestinationName(Session session, String destinationName,
2. boolean pubSubDomain) throws JMSException {
3. Assert.state(this.beanFactory != null, "BeanFactory is required");
4. try {
5. return (Destination) this.beanFactory.getBean(destinationName, Destination.class);
6. }
7. catch (BeansException ex) {
8. throw new DestinationResolutionException(
9. "Failed to look up Destinaton bean with name '" + destinationName + "'", ex);
10. }
11. }

2 JmsAccessor
抽象類JmsAccessor是JmsTemplate、SimpleMessageListenerContainer和 DefaultMessageListenerContainer等concrete class的基類。JmsAccessor定義瞭如下幾個用於訪問JMS服務的共通屬性。
Java程式碼
1. private ConnectionFactory connectionFactory;
2. private boolean sessionTransacted = false;
3. private int sessionAcknowledgeMode = Session.AUTO_ACKNOWLEDGE;

JmsAccessor提供了建立Connection和Session的方法，如下：
Java程式碼
1. protected Connection createConnection() throws JMSException {
2. return getConnectionFactory().createConnection();
3. }
4.
5. protected Session createSession(Connection con) throws JMSException {
6. return con.createSession(isSessionTransacted(), getSessionAcknowledgeMode());
7. }

2.1 JmsDestinationAccessor
抽象類JmsDestinationAccessor繼承自JmsAccessor，增加了destinationResolver和 pubSubDomain屬性。destinationResolver的預設值是DynamicDestinationResolver的例項，也就是 說預設採用動態目的地解析的方式；pubSubDomain用於指定是使用“釋出/訂閱”模式還是使用“點對點”模式，預設值是false。

JmsDestinationAccessor提供了用於解析目的地的方法，如下：
Java程式碼
1. protected Destination resolveDestinationName(Session session, String destinationName)
2. throws JMSException {
3. return getDestinationResolver().resolveDestinationName(session, destinationName,
4. isPubSubDomain());
5. }

2.2 AbstractJmsListeningContainer
AbstractJmsListeningContainer繼承自JmsDestinationAccessor，作為所有Message Listener Container的公共基類。它主要提供了JMS connection的生命週期管理的功能，但是沒有對訊息接收的方式（主動接收方式或者非同步接收方式）等做任何假定。該類主要的屬性如下：
Java程式碼
1. private String clientId;
2. private Connection sharedConnection;
clientId通常用於持久訂閱；sharedConnection儲存了被共享的JMS connection。

該類定義瞭如下的抽象方法，以便子類可以決定是否使用共享的JMS connection。
Java程式碼
1. protected abstract boolean sharedConnectionEnabled();

2.3 AbstractMessageListenerContainer
AbstractMessageListenerContainer繼承自AbstractJmsListeningContainer，也是作為所有Message Listener Container的公共基類。該類主要的屬性如下：
Java程式碼
1. private volatile Object destination;
2. private volatile Object messageListener;
3. private boolean exposeListenerSession = true;
destination用於指定接收訊息的目的地。
messageListener用於指定處理訊息的listener。對於messageListener，它既可以是符合JMS規範的 javax.jms.MessageListener，也可以是Spring特有的 org.springframework.jms.listener.SessionAwareMessageListener。 SessionAwareMessageListener的定義如下：
Java程式碼
1. public interface SessionAwareMessageListener {
2. void onMessage(Message message, Session session) throws JMSException;
3. }
跟javax.jms.MessageListener相比，這個介面的onMessage方法增加了一個Session 型別的引數，可以通過這個session傳送回覆訊息（reply message）。

如果使用了SessionAwareMessageListener 型別的message listener，那麼exposeListenerSession引數指定了傳入onMessage方法的session引數是否是建立了 MessageConsumer的session，預設值是true。如果是false，那麼 AbstractMessageListenerContainer會在connection上新建一個session，並傳入onMessage方法。

2.4 AbstractPollingMessageListenerContainer
AbstractPollingMessageListenerContainer繼承自AbstractMessageListenerContainer，它提供了對於主動接收訊息(polling)的支援，以及支援外部的事務管理。
Java程式碼
1. private boolean pubSubNoLocal = false;
2. private long receiveTimeout = DEFAULT_RECEIVE_TIMEOUT;
3. private PlatformTransactionManager transactionManager;
如果使用“釋出/訂閱”模式，那麼pubSubNoLocal 屬性指定通過某個連線傳送到某個Topic的訊息，是否應該被投遞迴這個連線。

receiveTimeout屬性用於指定呼叫MessageConsumer的receive方法時的超時時間，預設值是1秒。需要注意的是，這個值應該比transactionManager 中指定的事務超時時間略小。

通常情況下，應該為transactionManager設定一個 org.springframework.transaction.jta.JtaTransactionManager的例項，此外也要設定一個支援 XA的ConnectionFactory。需要注意的是，XA 事務對效能有較大的影響。
如果只是希望使用local JMS transaction，那麼只要設定sessionTransacted為true或者使用JmsTransactionManager即可。實際上， 如果設定了非JTA的transactionManager，那麼sessionTransacted屬性會自動被設定成true。
由於local JMS transaction無法同其它local transaction（例如local database transaction）進行協調，因此客戶端程式可能需要對重發的訊息進行檢查。JMS規範要求：JMS provider應該將重發訊息的JMSRedelivered屬性設定為true。

2.5 SimpleMessageListenerContainer
SimpleMessageListenerContainer繼承自AbstractMessageListenerContainer，使用非同步方式 接收訊息（也就是通過MessageConsumer上註冊MessageListener的方式接收訊息）。該類主要的屬性如下：
Java程式碼
1. private boolean pubSubNoLocal = false;
2. private int concurrentConsumers = 1;
3. private Set sessions;
4. private Set consumers;
5. private TaskExecutor taskExecutor;
如果使用“釋出/訂閱”模式，那麼pubSubNoLocal 屬性指定通過某個連線傳送到某個Topic的訊息，是否應該被投遞迴這個連線。

SimpleMessageListenerContainer允許建立多個Session和MessageConsumer來接收訊息。具體的個數由 concurrentConsumers屬性指定。需要注意的是，應該只是在Destination為Queue的時候才使用多個 MessageConsumer（Queue中的一個訊息只能被一個Consumer接收），雖然使用多個MessageConsumer會提高訊息處理 的效能，但是訊息處理的順序卻得不到保證：訊息被接收的順序仍然是訊息傳送時的順序，但是由於訊息可能會被併發處理，因此訊息處理的順序可能和訊息傳送的 順序不同。此外，不應該在Destination為Topic的時候使用多個MessageConsumer，這是因為多個 MessageConsumer會接收到同樣的訊息。
SimpleMessageListenerContainer建立的Session和MessageConsumer分別儲存在sessions和consumers屬性中。

taskExecutor屬性的預設值是null，也就是說，對MessageListener（或者 SessionAwareMessageListener）的回撥是在MessageConsumer的內部執行緒中執行。如果指定了 taskExecutor，那麼回撥是在TaskExecutor內部的執行緒中執行。以下是相關的程式碼：
Java程式碼
1. protected MessageConsumer createListenerConsumer(final Session session)
2. throws JMSException {
3. Destination destination = getDestination();
4. if (destination == null) {
5. destination = resolveDestinationName(session, getDestinationName());
6. }
7. MessageConsumer consumer = createConsumer(session, destination);
8.
9. if (this.taskExecutor != null) {
10. consumer.setMessageListener(new MessageListener() {
11. public void onMessage(final Message message) {
12. taskExecutor.execute(new Runnable() {
13. public void run() {
14. processMessage(message, session);
15. }
16. });
17. }
18. });
19. }
20. else {
21. consumer.setMessageListener(new MessageListener() {
22. public void onMessage(Message message) {
23. processMessage(message, session);
24. }
25. });
26. }
27.
28. return consumer;
29. }
需要注意的是，如果指定了taskExecutor，那麼訊息在被taskExecutor內部的執行緒處理前，可能已經被確認過了（外層的 onMessage方法可能已經執行結束了）。因此如果使用事務Session或者Session.CLIENT_ACKNOWLEDGE型別的確認模 式，那麼可能會導致問題。

該類的sharedConnectionEnabled方法（在AbstractJmsListeningContainer中定義）總是返回true， 因此SimpleMessageListenerContainer會使用共享的JMS connection。

2.6 DefaultMessageListenerContainer
DefaultMessageListenerContainer繼承自 AbstractPollingMessageListenerContainer，主要使用同步的方式接收訊息（也就是通過迴圈呼叫 MessageConsumer.receive的方式接收訊息）。該類主要的屬性如下：
Java程式碼
1. private int concurrentConsumers = 1;
2. private int maxConcurrentConsumers = 1;
3. private int maxMessagesPerTask = Integer.MIN_VALUE;
4. private int idleTaskExecutionLimit = 1;
5. private final Set scheduledInvokers = new HashSet();
6. private TaskExecutor taskExecutor;
7. private int cacheLevel = CACHE_AUTO;
跟SimpleMessageListenerContainer一樣，DefaultMessageListenerContainer也支援建立多個 Session和MessageConsumer來接收訊息。跟SimpleMessageListenerContainer不同的 是，DefaultMessageListenerContainer建立了concurrentConsumers所指定個數的 AsyncMessageListenerInvoker（實現了SchedulingAwareRunnable介面），並交給 taskExecutor執行。

maxMessagesPerTask屬性的預設值是Integer.MIN_VALUE，但是如果設定的taskExecutor（預設值是 SimpleAsyncTaskExecutor）實現了SchedulingTaskExecutor介面並且其 prefersShortLivedTasks方法返回true（也就是說該TaskExecutor傾向於短期任務），那麼 maxMessagesPerTask屬性會自動被設定為10。
如果maxMessagesPerTask屬性的值小於0，那麼AsyncMessageListenerInvoker.run方法會在迴圈中反覆嘗試 接收訊息，並在接收到訊息後呼叫MessageListener（或者SessionAwareMessageListener）；如果 maxMessagesPerTask屬性的值不小於0，那麼AsyncMessageListenerInvoker.run方法裡最多會嘗試接收訊息 maxMessagesPerTask次，每次接收訊息的超時時間由其父類 AbstractPollingMessageListenerContainer的receiveTimeout屬性指定。如果在這些嘗試中都沒有接收 到訊息，那麼AsyncMessageListenerInvoker的idleTaskExecutionCount屬性會被累加。在run方法執行完 畢前會對idleTaskExecutionCount進行檢查，如果該值超過了 DefaultMessageListenerContainer.idleTaskExecutionLimit（預設值1），那麼這個 AsyncMessageListenerInvoker可能會被銷燬。

所有AsyncMessageListenerInvoker例項都儲存在scheduledInvokers中，例項的個數可以在 concurrentConsumers和maxConcurrentConsumers之間浮動。跟 SimpleMessageListenerContainer一樣，應該只是在Destination為Queue的時候才使用多個 AsyncMessageListenerInvoker例項。

cacheLevel屬性用於指定是否對JMS資源進行快取，可選的值是CACHE_NONE = 0、CACHE_CONNECTION = 1、CACHE_SESSION = 2、CACHE_CONSUMER = 3和CACHE_AUTO = 4。預設情況下，如果transactionManager屬性不為null，那麼cacheLevel被自動設定為CACHE_NONE（不進行緩 存），否則cacheLevel被自動設定為CACHE_CONSUMER。

如果cacheLevel屬性值大於等於CACHE_CONNECTION，那麼sharedConnectionEnabled方法（在AbstractJmsListeningContainer中定義）返回true，也就是說使用共享的JMS連線。


3 SingleConnectionFactory
SingleConnectionFactory實現了ConnectionFactory介面，其createConnection方法總是返回相同的 Connection。可以在SingleConnectionFactory的建構函式中傳入Connection物件或者 ConnectionFactory物件，用來建立被代理的連線物件。 SingleConnectionFactory.createConnection方法返回的連線是個代理，它忽略了對stop和close方法的呼叫 （連線會在SingleConnectionFactory.destroy方法中關閉）。

SingleConnectionFactory的reconnectOnException屬性用來指定是否在連線丟擲JMSException的時候，對連線進行重置，重置後如果再呼叫createConnection方法，那麼會返回一個新的連線。

需要注意的是，AbstractJmsListeningContainer類的抽象方法sharedConnectionEnabled指定了是否在 message listener container內部使用共享的JMS連線。因此通常情況下不需要為單獨的message listener container設定SingleConnectionFactory（及其子類）；如果希望在不同的message listener container之間共享JMS連線，那麼可以考慮使用SingleConnectionFactory。

3.1 CachingConnectionFactory
CachingConnectionFactory繼承自SingleConnectionFactory，增加了對Session和MessageProducer快取的功能。該類主要的屬性如下：
Java程式碼
1. private int sessionCacheSize = 1;
2. private boolean cacheProducers = true;
sessionCacheSize屬性指定了被快取的Session例項的個數（預設值是1），也就是說，如果同時請求的Session個數大於sessionCacheSize，那麼這些Session不會被快取，而是正常的被建立和銷燬。

cacheProducers屬性指定了是否對MessageProducer進行快取，預設值是true。