
JMS簡介:一種應用于異步消息傳遞的標準API,JMS也是應用于程序間通訊的。但是,JMS與其他機制有所不同,主要表現在系統間傳遞信息的方式,見PPT1-2。簡介傳送也是JMS的關鍵。當一個應用程序通過JMS向另一個應用程序發送消息時,兩個程序之間并沒有直接的連接。發送應用程序會將消息交給一個服務,由服務確保將消息投遞給接收應用程序。在JMS中有兩個主要的概念:消息中介(message broker)和消息目標(destination)。當應用程序發送消息時,會將消息交給一個消息中介。消息中介實際上就是JMS版本的郵局。消息中介可以確保消息被投遞到指定的消息目標,同時可以釋放發送者,使其能夠進行其他的業務。 在JMS中,每條消息帶有一個消息目標。消息目標就好像一個郵箱,可以將消息放入這個郵箱,直到有人將它們取出。消息目標只關心消息應該從哪里獲得--而不是由誰來獲得。在JMS中,有兩種消息目標類型:隊列和主題。分別應用于隊列的點對點模型或應用于主題的發布--訂閱模型。
-
點對點消息傳遞模型:在點對點模型中,每個消息都有一個發送者和一個接收者,見PPT3.在JMS中如果有多個接收者監聽隊列,就沒辦法直到某條特定的消息會被哪個接收者處理。這種不確定性實際上有很多好處,因為它可以讓程序之用為隊列添加監聽器就能增大消息處理的能力。
發布--訂閱消息傳遞模型:在此模型中,消息會被發送給一個主題。像使用隊列一樣,可以讓多個接收者監聽一個主題。但是,與隊列不同的是消息不再被只投遞給一個接收者,所有主題的訂閱者都會收到消息,見PPT4。
JMS的優點:不用等待--當使用JMS發送消息時,客戶端不必等待消息被處理,甚至是被投遞。客戶端只需要將消息發給消息中介,就可以確信消息會被投遞到適當的目標。因為不必等待,客戶端就可以執行其他的任務。由于這種方法可以大大地節省時間,客戶端的性能能夠極大地提高。面向消息--與RFC通信面向方法調用不同,使用JMS發送消息是以數據位中心的。這意味著客戶端不用固定搭配特定的方法符號。任何隊列或主題訂閱者都可以處理由客戶端發送來的消息。客戶端不必了解服務的任何規范。位置獨立--JMS客戶端不必知道由誰來處理他們的消息,或者服務的位置在哪里。客戶端值需要了解需要通過哪個隊列或主題發送消息。因此,只要能夠從隊列或主題獲取消息,JMS客戶端就不用在意服務的位置在哪里。確保投送--當使用JMS發送消息時,客戶端能夠確保消息被投遞。即使在消息發送時服務無法使用,消息也會被儲存起來,直到服務重新可以使用為止。
在Spring中安裝ActiveMQ:它是一個開源消息中介(Apache的一個子項目),也是應用JMS異步消息傳遞的很好選擇。需要從www.activemq.org下載包。在bin目錄中,可以找到一個用于啟動ActiveMQ的腳本:Unix用戶的activemq或者Windows用戶的activemq.bat。運行這個腳本,等ActiveMq啟動后,就可以使用它進行中介服務了。在所有的示例中,都需要通過JMS連接工廠通過消息中介發送消息。我們選擇了ActiveMQ作為消息中介,因此必須配置JMS連接工廠,使它能夠了解如何連接到ActiveMQ,ActiveMQConnectionFactory是連接ActiveMQ的JMS連接工廠,在Spring中配置如下:
<bean id ="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory" >
<property name="brokerURL" value="tcp://localhost:61616">
</bean>
-
聲明一個ActiveMQ隊列:
<bean id="rantzDestination" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg index="0" value="rantz.marketing.queue" />
</bean>
聲明一個ActiveMQ主題:
<bean id="rantzDestination" class="org.apache.activemq.command.ActiveMQTopic">
<constructor-arg index="0" value="rantz.marketing.topic" />
</bean>
協同使用JMS和Spring:JMS為Java愛好者提供了一個與消息中介進行交互,以及發送和接收消息的標準API,而且每一個消息中介的實現都會支持JMS。因此你不必由于消息中介的不同而學習不同的消息傳遞的API。雖然JMS為所有的消息中介提供了統一接口,但這種接口用起來不是十分便利(見PPT5-6)
-
使用JMS模板:JmsTemplate是Spring消除冗長和重復JMS代碼的解決方案。JmsTemplate可以創建連接,獲取會話,以及發送和接收消息。它使你可以專注于構建要發送的消息或者處理收到的消息。另外,JmsTemplate可以處理任何被拋出的JMSException。如果JMSException在JmsTemplate工作中被拋出,JmsTemplate將捕獲這個異常,并且用一個未受查的JmsException的子類再次拋出它。見PPT7
置入JmsTemplate模板:
<bean id ="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory" ref="connectionFactory">
</bean>
-
發送消息:在RoadRantz端,我們使用JmsTemplate向RoandRantz市場發送駕駛員信息。RantzMarketingGatewayImpl是RoadRantz與市場系統進行交互的類。見PPT8,因此,當我們在Spring中配置RantzMarKetingGatewayImpl類時,必須置入jmsTemplate和rantzDestination Bean的參考:
<bean id="marketingGateway" class="com.roadrantz.marketing.RantzMarketingGatewayImpl">
<property name="jmsTemplate" ref ="jmsTemplate" />
<property name="destination" ref="rantzDestination" />
</bean>
設置默認目標:如果每次發送消息都指定一個目標,不如為JmsTemplate置入一個默認目標:
<bean id ="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory" ref="connectionFactory" />
<property name="defaultDestination" ref="rantzDestination" />
</bean>
現在,調用JmsTemplate的send()方法就可以去掉第一個參數了。這種send()方法的形式只帶有一個MessageCreator。因為沒有指定目標,JmsTemplate會假設你要將消息發送給默認目標。所以不必再為RantzMarketingGatewayImpl注入目標了。它的聲明很簡單:
<bean id="marketingGateway" class="com.roadrantz.marketing.RantzMarketingGatewayImpl">
<property name="jmsTemplate" ref ="jmsTemplate" />
</bean>
消費消息:使用JmsTemplate接收消息十分簡單。只需要調用JmsTemplate的receive()方法,如PPT9. 默認情況下,對receive()方法的調用會造成阻塞,直到消息到達目標--如果必要,會永遠等待下去。為了避免對消息的內部等待,可以通過配置JmsTemplate時,通過設置receiveTimeout屬性來指定接收時間。下面配置的接收超時時間為一分鐘(60000毫秒)。
<bean id ="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory" ref="connectionFactory" />
<property name="defaultDestination" ref="rantzDestination" />
<property name="receiveTimeout" value="60000" />
</bean>
在PPT9中,receive()方法會從默認目標接收消息。如果你希望指定一個目標,可以為其傳遞一個目標:
MapMessage message = (MapMessage) jmsTemplate.receive(destination);
此外,還可以通過名稱指定目標,并且讓Spring的目標解析器自動解析這個目標:
MapMessage message = (MapMessage) jmsTemplate.receive("rantz.marketing.queue");
轉換消息:為了簡化示例,沒有在發送和接收消息的代碼中添加處理消息轉換的代碼。但是,如果在程序中的多個位置都需要發送或接收相同的消息,可能就需要通過消息轉換器來避免不必要的映射代碼。盡管將消息轉換代碼抽象成自己的工具類并不是很困難,但是你仍然需要明確的調用工具類來進行轉換。幸運的是,Spring通過MessageConverter接口提供了對消息轉換的支持:
public interface MessageConverter {
public Message toMessage(Object object, Session session);
public Object fromMessage (Message message) ;
}
見PPT10,展示了一個MessageConverter接口的實現MotoristMessageConverter,可以將Motorist對象轉換為消息,也可以將消息轉換為SpammedMotorist對象。
-
發送和接收被轉換的消息:我們可以在發送消息前不用明確的調用toMessage()方法,只需要調用JmsTemplate的convertAndSend()方法。因此,PPT8中的sendMotoristInfo()方法將變得更加簡單:
public void sendMotoristInfo (final Motorist motorist){jmsTemplate.converAndSend(motorist); } 消息被發向JmsTemplate的默認目標(假設已經指定了一個默認目標)。不過我們也可以在調用convertAndSend()方法時指定一個特定的目標: jmsTemplate.connverAndSend(destination,motorist); 另外,我們也可以通過名稱指定目標: jmsTemplate.convertAndSend("rantz.marketing.queue",motorist); 接收端: public SpammedMotorist receiveSpammedMotorist() { return (SpammedMotorist) jmsTemplate.receiveAndConvert(); } 除非有特殊的指定,recieveAndConvert()會從默認目標接收消息。不過,我們也可以通過為receiveAndConvert()方法傳遞一個參數來選擇一個目標: return (SpammedMotorist) jmsTemplate.receiveAndConvert(destination); 或者使用目標的名稱: return (SpammedMotorist) jmsTemplate.receiveAndConvert("rantz.marketing.queue");置入消息轉化器: <bean id="motoristConverter" class="com.roadrantz.marketing.MotoristMessageConverter" />
最后,JmsTemplate需要了解這個消息轉換器。為了提供消息轉換器,我們會將motoristConverter Bean置入到JmsTemplate的messageConverter屬性中:
<bean id ="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory" ref="connectionFactory" />
<property name="defaultDestination" ref="rantzDestination" />
<property name="messageConverter" ref="motoristConverter" />
</bean>
-
將Spring的網關支持類應用于JMS: Spring通過提供JdbcDaoSupport使JdbcTemplate的應用比原先變得更簡單一些。與此類似,Spring也提供了一個用于JMS網關類的基類JmsGatewaySupport(見PPT11)。 那么JmsGatewaySupport從哪里得到JmsTemplate呢?你可以直接在jmsTemplate屬性中注入JmsTemplate,就像處理常規的RantzMarketingGatewayImpl一樣。或者,可以為connectionFactory屬性置入連接工廠,來滿足對JmsTemplate Bean的所有需求:
<bean id = "marketingGateway" class="com.roadrantz.marketing.RantzMarketingGatewayImple">
<property name="connectionFactory " ref="connectionFactory" />
</bean>
當這種方法配置時,JmsGateSupport將基于被注入的連接工廠自動創建一個JmsTemplate對象。因此不必在Spring中聲明JmsTemplate Bean。 在直接將連接工廠置入到網關之前,你應該了解這種方法有兩個缺點:
1、只能在JmsTemplate上指定默認目標。如果JmsGateSupport創建了自己的JmsTemplate,就沒有機會再指定默認的目標了。你必須在調用send()或receive()時明確地選擇一個目標。
2、只能將消息轉換器置入到JmsTemplate中。如果JmsGatewaySupport創建了自己JmsTemplate,將不能使用消息轉換器。因此,必須明確地在網關代碼中處理消息的轉換。
-
創建消息驅動POJO:EJB2.0規范的其中一個重要內容是包含了消息驅動Bean(MDB)。MDB是可以異步處理消息的EJB。換句話說,MDB會將JMS目標中的消息作為事件來響應。這與同步消息接收者在消息可用前進行阻塞正好相反。MDB是EJB的一個兩點。在EJB3的規范中,MDB被簡化了,使其更像POJO。不再需要實現MessageDrivenBean接口,只需要實現常規的java.jms.MessageListener接口并用@MessageDriven注釋MDB。Spring2.0通過自己的消息驅動Bean形式來滿足消息的異步消費需求。
-
創建消息監聽器:MarketingMdp不必實現MessageDrivenBean接口的世界是多么簡單,見PPT12。MarketingMdp本身并不做些什么。它有一個實際處理消息的onMessage()方法。不過要先在Spring中配置一下:
<bean id="rantzMdp" class="com.roadrantz.marketing.MarketingMdp" />
EJB3.0MDB會使用@MessageDriven注釋通知容器這是一個MDB。但是,在Spring中,我們會通過將其注入到一個消息監聽容器來指示這個Bean是一個MDP。
包含消息監聽器:消息監聽器容器是一個用于查看JMS目標,等待消息到達的特殊Bean。一旦消息到達,它就可以獲取到消息,并通過onMessage()方法將消息傳遞給一個MessageListener的實現。因為MarketingMdp類實現了MessageListener接口,所以消息監聽器容器就準備完畢了。見PPT13. 顧名思義,SimpleMessageListenerContainer是最簡單的消息監聽器容器,可以按下面的方法在Spring中進行配置:
<bean class="org.springframework.jms.listener.SimpleMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory" />
<property name="destination" ref="rantzDestination" />
<property name="messageListener" ref="rantzMdp" />
</bean>
對于messageListener屬性,我們為其置入了對MDP實現的引用,這樣,onMessage()方法將可以被用來接收消息。
使用事務性的MDP:如果收到的一個消息在事務中,則應該使用DefaultMessageListenerContainer:
<bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory" />
<property name="destination" ref="rantzDestination" />
<property name="messageListener" ref="rantzMdp" />
<property name="transactionManager" ref="jmsTransactionManager" />
</bean> 如果事務性需求比較簡單,JmsTransactionManager將按如下方法配置:
<bean id="jmsTransactionManager" class="org.springframework.jms.connection.JmsTransactionManager">
<property name="connectionFactory" ref="connectionFactory" />
</bean> 需要提醒的是,tansactionManger屬性是可選的。如果不注入事務管理器,MDP就不是事務性的。
編寫純POJO MDP:如果消息監聽器容器的messageListener屬性被注入了MessageListener的實現,它就能夠知道在消息到達時應該調用onMessage()方法。幸運的是,Spring提供了一個替代的MessageListenerAdapter。它是一個MessageListener,可以委派Bean和你選擇的方法,見PPT14. 如果不將自己的MessageListener的實現注入到消息監聽器容器中,你可以置入到MessageListenerAdapter中:
<bean class="org.springframework.jms.listener.SimpleMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory" />
<property name="destination" ref="rantzDestination" />
<property name="messageListener" ref ="purePojoMdp" />
</bean> 因為配置了purePojoMdp Bean,所以它是一個MessageListenerAdapter:
<bean id="purePojoMdp" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
<property name="delegate" ref="rantzMdp" />
<property name="defaultListenerMethod" value="processMotoristInfo" />
</bean> 默認情況下,MessageListenerAdapter在消息到達時會調用handleMessage()方法。但是,我們希望MarketingMdp Bean可以通過 processMotoristInfo()方法處理消息,因此將defaultListenerMethod設置為processMotoristInfo. 因為選擇了一個特定的被調用方法,所以不需要實現MessageListener或onMessage()方法。因此 MarketingMdp現在將被簡化為PPT15。盡管它是一個POJO,對MapMessage的依賴造成了MarketingMdp與JMS的不必要耦合,另外,MapMessage的getString方法還會拋出必須被處理的JMSException。理想情況下,MarketingMdp不應該依賴任何特定框架的類型。 當MessageListenerAdapter接收消息時,它會考慮消息的類型和defaultListenerMethod的值,并且嘗試著查找用來調用的監聽器方法符號。PPT16描述了MessageListenerAdapter是如何將JMS消息映射到監聽器方法參數的。
轉換MDP消息:在最新版中,processMotoristInfo()帶有的是Map,并且在處理前需要將Map轉換為SpammedMotorist。如果在消息到達時,能夠直接給processMotoristInfo()方法傳遞可以處理的SpammedMotorist對象豈不是更好么?Spring消息轉換器可以執行消息和特定域Java類型之間的相互轉換工作。在PPT10中,已經有一個消息轉換器。我們需要做的就是讓MessageListenerAdapter能夠感知這個消息轉換器。MessageListenerAdapter的messageConverter屬性可以完成這項工作:
<bean id="purePojoMdp" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
<property name="delegate" ref="rantzMdp" />
<property name="defaultListenerMethod" value="processMotoristInfo" />
<propert name="messageConverter" ref="motoristConverter" />
</bean> 現在,可以編寫最終版的MarketingMdp了。
使用基于消息的RPC:RPC編程模型可以使與遠程服務的交互就如同在調用本地對象的方法。是否有某種方法既具有RPC編程模型的簡單性,又可以利用異步消息傳遞的優點呢?
-
引入Lingo: 它是一種基于Spring的遠程調用方法,它在RPC和異步消息傳遞之間架起了一座橋梁。與使用其他的Spring遠程調用方法相同,Lingo提供了一個服務輸出器,可以將Bean的功能輸出為Lingo服務和客戶端代理。 Lingo遠程調用會通過JMS隊列或主題來承載信息。 盡管Lingo是基于Spring遠程調用的,但是它并不是Spring框架的一部分。你可以從Lingo的主頁下載Lingo(http://lingo.codehaus.org/Download).
輸出服務:在服務端,Lingo提供了一個服務輸出器JmsServiceExporter。被Lingo輸出的服務是通過JMS來提供服務的,而不是用于直接的RPC訪問,見PPT17。下面的XML配置了一個JmsServiceExporter,能夠將rantzMdp Bean輸出為RPC-over-JMS服務:
<bean id="server" class="org.logicblaze.lingo.jms.JmsServiceExporter">
<property name="connectionFactory" ref="connectionFactory" />
<property name="destination" ref="destination" />
<property name="service" ref="rantzMdp" />
<property name="serviceInterface" value="com.roadrantz.marketing.MarketingService" />
</bean>
service屬性被置入了對rantzMdp Bean的引用,即MarketingMdp。最后,應該將定義了服務的接口的類名配置到serviceInterface屬性中。我們聲明的被輸出服務帶有MarketingService接口,定義如下:
public interface MarketingService {
void processMotoristInfo (SpammedMotorist motorist );
} 因為我們定義的服務帶有MarketingService接口,這就意味著我們應該對MarketingMdp類進行一些小的修改,以便它可以實現MarketingService接口:
public class MarketingMdp implements MarketingService {
public void processMotoristInfo(SpammedMotorist motorist) {
...
}
}
這就是使用Lingo輸出服務的全過程。一旦應用程序啟動,JmsServiceExporter將會起作用,就可以開始使用它了。現在,讓我們再到客戶端看看RoadRantz應用程序是如何調用這個被輸出的市場服務的。
代理JMS:在RoadRantz應用程序中,每次用戶注冊并選擇愿意接收特定提供商信息時都需要調用processMotoristInfo()方法,因此,我們必須以某種方法將Lingo輸出服務的引用置入到RoadRantz應用程序中。
-
置入JmsProxyFactoryBean:lingo提供了JmsProxyFactoryBean,這是一個代理工廠Bean,可以生成遠程Lingo輸出服務的代理。如PPT18所示,通過JmsProxyFactoryBean代理的服務是通過JMS目標(隊列或主題)訪問的,而不是通過TCP/IP。下面聲明配置了一個JmsProxyFactoryBean:
<bean id = "marketing" class="org.logicblaze.lingo.jms.JmsProxyFactoryBean">
<property name="connectionFactory" ref="connectionFactory" />
<property name="destination" ref="destination" />
<property name="serviceInterface" value="com.roadrantz.marketing.MarketingService" />
</bean> serviceInterface屬性指定了代理需要實現的Java接口。通過這個接口,RoadRantz可以調用processMotoristInfo()方法。 對于配置JmsProxyFactoryBean,最需要注意的事情是不用配置服務的位置信息。這是因為服務的位置并不重要,只需要知道服務在哪里接收"郵件"。事實上,我們還可以使用遠程服務的多個實例。如果想建立一個高可靠性的市場服務,可以啟動兩個或多個實例,讓它們都監聽相同的目標。每一個實例都能夠處理一個請求。其間,客戶端并不知道有一個服務池在等待請求進行響應。
進行調用:在JmsProxyFactoryBean被置入后,就可以開始調用遠程服務了。我們需要做的是將其置入到RantServiceImpl:
<bean id ="rantService" class="com.roadrantz.service.RantServiceImpl">
<property name="rantDao" ref="rantDao" />
<property name="marketingService" ref="marketing" />
</bean> 接著,我們可以使用它在addMotorist()方法中向市場服務發送SpammedMotorist對象。PPT19展示了為了調用遠程市場服務對RantServiceImpl進行的相關修改。 正如你看到的,調用Lingo輸出的服務與調用RMI服務、Web服務或在同一個進程中調用其他Bean上的方法是相同的。PPT19中沒有任何關于JMS的內容。唯一不同的地方在Spring的配置。通過這種方法,只需要簡單的更改Spring的配置就能夠在JMS和其他通信機制直接進行切換。
-
-- 學海無涯