網絡拓撲 發布者 再看 不讓 div 模式 cal nta 封裝

這一節來說說，異步機制及spring對JMS封裝

一、消息異步處理

類似於RMI、Hessian、Burlap等遠程方法調用，它們都是同步的，所謂同步調用就是客戶端必須等待操作完成，如果遠程服務沒有返回任何響應，客戶端會一直等待直到服務完成。

所謂同步：就是客戶端必須等待操作完成，如果遠程服務沒有任何返回響應，客戶端會一直等待直到服務完成。

異步調用：客戶端發送消息無需等待服務處理完成便可立即返回，就像發送完消息就立刻被處理成功一樣。

1.1消息的 發送

異步處理的世界，我們可以把消息發送比作一個郵局系統。我們不必關心郵件如何送出，能否到達，郵局系統會保證郵件最終送達我們希望的接受者手中，點對點 和發布、訂閱。

1.1點對點模式

在點對點模式中，每個消息只有一個發送者和一個接受者，

在點對點模型中，每個消息只有一個發送者和一個接收者。如下圖所示：

在點對點模型中， 消息broker會把消息放入一個queue。當一個接收者請求下一個消息時，消息會被從queue中取出並傳遞給接收者。因為消息從queue中取出便會被移除，所以這保證了一個消息只能有一個接收者。

盡管消息隊列中的每個消息只有一個接收者，但這並不意味著只能有一個接收者從隊列獲取消息，可以同時有多個接收者從隊列獲取消息，只不過它們只能處理各自接收到的消息。其實這就像在銀行排隊一樣，排隊的人可以看做一個個消息，而銀行工作窗口便是消息的接收者，每個窗口服務完一個客戶之後都會讓隊列中的“下一個”到窗口辦理業務。

還有，如果多個接收者監聽一個隊列，我們是很難確定到底哪個接收者處理哪個消息的。不過這也不一定不好，因為這樣就使得我們很方便的通過增加接收者來拓展應用處理能力了。

1.2發布/訂閱模式

在發布/訂閱模式中，消息是被發送到topic中的。就像queue一樣，很多接收者可以監聽同一個topic，但是與queue每個消息只傳遞給一個接收者不同，訂閱了同一個topic的所有接收者都會收到消息的拷貝，如下圖所示：

從發布/訂閱的名字中我們也可看出，發布者發布一條消息，所有訂閱者都能收到，這就是發布訂閱模式最大的特性。對於發布者來說，它只知道將消息發布到了一個特定的topic，它不關心誰監聽這個topic，這也就意味著它並不知道這些消息是被如何處理的。

1.2 異步消息系統帶來的好處

在具體介紹異步消息系統帶來的好處之前，我們先看看同步系統的局限性：

同步會話意味著等待：當客戶的調用遠程服務的方法時，客戶端必須等待遠程方法結束之後才能繼續，如果客戶端與遠程服務交流頻繁或者遠程服務響應過慢，會影響客戶端的性能 客戶端和服務接口耦合：如果服務接口發生改變，所有客戶的都需要修改 客戶端和服務位置耦合：客戶端要想使用遠程服務就必須配置服務的地址，如果網絡拓撲發生變化，客戶端需要重新配置服務地址 客戶端和服務可用性耦合：如果服務不可用，那麽也會導致客戶端不可用

下面我們再看一下異步消息系統是如何解決這些問題的。

無需等待

當一個消息被異步發送，客戶端不需要等待它處理完成。客戶端直接把消息扔給broker然後做其它事情，broker負責把消息送到合適的目的地。

因為客戶端不需要等待，所以客戶端的性能會有很大的提升。

面向消息和解耦合

不同於傳統基於方法調用的RPC會話，消息異步發送是以數據為中心的。這就意味著客戶端不需要和某個方法簽名綁定，任何queue或topic的訂閱者都可以處理客戶端發送的消息。客戶端不必再關心服務方任何相關的問題。

位置獨立

同步RPC服務的調用是通過網絡地址定位的，這就意味著客戶端無法擺脫網絡拓撲的變化。如果服務的IP或端口發生改變，客戶端也需要做相應的改變。

相反，異步消息系統中的客戶端並不關心服務所在的位置及其如何處理消息，它只負責將消息發送到特定的queue或topic。所以，服務位於什麽地方都無所謂，只要它們能夠從queue或topic中獲取消息即可。

在點對點模式中，可以很方便的利用位置獨立這個特性創建一個服務集群。客戶端不需要關心服務的位置，集群中各個服務僅需知道broker的位置，並從同一個queue獲取消息，如果服務壓力過大無法及時處理消息，我們只需要在集群中增加一個服務實例去監聽同一個queue即可。

在發布/訂閱模式中，位置獨立同樣有很重要的作用。多個服務可以訂閱同一個topic，他們都能獲取到topic中的每個消息，但是對各個服務的處理可以不同。比如我們有一個服務集合訂閱了一個接收新員工消息的topic，所以這些服務都可以得到每個新員工消息，一個服務可以將新員工添加到薪資系統，另一個服務可以將新員工增加到hr系統，還有服務負責賦予新員工各種系統權限等等，每個訂閱topic的服務都能對各自的消息做出自己的處理。

可靠性保證

當一個客戶端和服務通過同步方式進行交互時，如果服務出現任何問題掛掉，都會影響客戶端正常工作。但是當消息是異步發送時，客戶端與服務之間被broker隔離，客戶端只負責發送消息，即使當發送消息時服務掛掉，消息也會被broker存儲起來，等到服務可用時再接著進行處理。

二、通過JMS發送消息

Java Message Service是一個Java標準，它定義了一套與消息broker交互的通用API。在JMS出現之前，每一種消息broker都有自己獨特的一套API，使得應用代碼無法在不同的broker之間適用。但是通過JMS，所有與broker交互的代碼就可以適用一套通用的API，就像JDBC一樣。

當然Spring對JMS也提供了支持，即JmsTemplate。通過JmsTemplate，我們可以更加方便地向queue和topic發送和接收消息。後面我們會詳細介紹Spring對JMS的實現，但是在發送和接收消息之前，我們需要現有一個broker。

2.1 在Spring中配置消息broker

ActiveMQ是非常優秀的JMS框架，關於ActiveMQ相關內容這裏不多做介紹，具體可以參考：http://activemq.apache.org/，本篇主要介紹如何在Spring中對其進行配置和使用。

2.1.1 創建一個connection factory

我們要想發送消息到ActiveMQ，就需要先創建到它的連接，ActiveMQConnectionFactory就是JMS中負責創建到ActiveMQ連接的工廠類。在Spring中配置方式如下：

?

1 2	<code><bean class="org.apache.activemq.spring.ActiveMQConnectionFactory" id="connectionFactory" p:brokerurl="tcp://localhost:61616"> </bean></code>

除此之外，Spring為ActiveMQ提供了專門的命名空間，我們可以使用Spring的ActiveMQ命名空間來創建連接工廠。首先要在配置文件中聲明amq命名空間：

?

1 2 3 4 5 6 7 8 9 10

<code><!--?xml version="1.0" encoding="UTF-8"?--> <beans xmlns="http://www.springframework.org/schema/beans" xmlns:amq="http://activemq.apache.org/schema/core" xmlns:jms="http://www.springframework.org/schema/jms" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemalocation="http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd http://www.springframework.org/schema/jms http://www.springframework.org/schema/jms/spring-jms.xsd http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd"> ... </beans> </code>

然後我們就可以利用元素來聲明一個連接工廠：

?

1 2	<code> </amq:connectionfactory></code>

需要註意，元素是專門針對ActiveMQ的。如果我們用到的是其它broker，就需要用另外的標簽元素或註入另外的工廠bean。上面元素中的brokerURL指定了ActiveMQ在服務器中的IP和端口，上面端口值就是ActiveMQ默認端口。

2.1.2 聲明ActiveMQ的消息destination

除了要有一個連接工廠之外，我們還需要知道消息發送到的destination。上面講過了，消息的destination只有兩類queue或者topic，在Spring中，我們需要配置queue或topic對應的bean。

配置一個ActiveMQ queue bean：

?

1 2	<code><bean c:_="biz1.queue" class="org.apache.activemq.command.ActiveMQQueue" id="queue"> </bean></code>

配置一個ActiveMQ topic bean：

?

1 2	<code><bean c:_="biz1.topic" class="org.apache.activemq.command.ActiveMQTopic" id="topic"> </bean></code>

上面例子中c:_屬性代表的是構造器參數，它指定了queue或topic的名稱。

像連接工廠一樣，Spring提供了另外一種配置destination的方式，就是通過Spring ActiveMQ命名空間進行配置。

使用元素配置一個queue：

?

1 2	<code> </amq:queue></code>

使用元素配置一個topic:

?

1 2	<code> </amq:topic></code>

上面元素中physicalName屬性代表消息通道的名稱，也就是queue和topic的名稱。

通過上面兩個組件的配置，我們就可以向ActiveMQ發送和接收消息了。發送和接收消息我們使用的是Spring提供的JmsTempate，它是Spring對JMS的抽象，下面就詳細介紹JMSTemplate的使用。

2.2 使用Spring的JMS template

雖然JMS提供了一套與各種broker交互的通用API，但實際使用起來並不是很方便，我們先看一下使用普通JMS API與broker交互的代碼。

2.2.1 通過普通JMS API發送消息到broker代碼：

?

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25

<code>ConnectionFactory cf =new ActiveMQConnectionFactory("tcp://localhost:61616"); Connection conn = null; Session session = null; try { conn = cf.createConnection(); session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE); Destination destination = new ActiveMQQueue("spitter.queue"); MessageProducer producer = session.createProducer(destination); TextMessage message = session.createTextMessage(); message.setText("Hello world!"); producer.send(message); } catch (JMSException e) { // handle exception? } finally { try { if (session != null) { session.close(); } if (conn != null) { conn.close(); } } catch (JMSException ex) { } } </code>

上面代碼中我們可以看到，為了發送一條 “Hello world”的消息卻用了20多行代碼，就像JDBC一樣，我們大部分代碼都是再做一些重復性的準備工作，比如獲取連接、創建session、異常處理等等。其實接收消息的代碼也是如此，在JDBC中，Spring提供了一個JdbcTemplate來簡化JDBC代碼開發，同樣，Spring也提供了JmsTemplate來簡化JMS消息處理的開發。

2.2.2 使用JmsTemplate

JmsTemplate其實是Spring對JMS更高一層的抽象，它封裝了大部分創建連接、獲取session及發送接收消息相關的代碼，使得我們可以把精力集中在消息的發送和接收上。另外，JmsTemplate對異常也做了很好的封裝，其對應的基本的異常為JMSException。

要使用JmsTemplate，就要在Spring配置文件中配置它作為一個bean：

?

1 2	<code><bean c:_-ref="connectionFactory" class="org.springframework.jms.core.JmsTemplate" id="jmsTemplate"> </bean></code>

因為JmsTemplate需要先和broker進行連接，所以它需要依賴一個connectionFactory。

因為JmsTemplate需要先和broker進行連接，所以它需要依賴一個connectionFactory。

發送消息

假如我們有一個業務需要用到異步消息發送，我們先定義這樣一個業務接口：

?

1 2 3 4

<code>public interface MyMessageService { void sendMessage(String message); } </code>

上面接口中只有一個方法，就是發送消息。

我們寫這個接口的實現，在這個接口實現中，我們就是用JmsTemplate實現異步消息發送：

?

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25

<code>import org.springframework.beans.factory.annotation.Autowired; import org.springframework.jms.core.JmsOperations; import org.springframework.jms.core.MessageCreator; import org.springframework.stereotype.Component; import javax.jms.JMSException; import javax.jms.Message; import javax.jms.Session; /** * Created by [email protected] on 2016/6/17. */ @Component public class MyMessageServiceImpl implements MyMessageService{ @Autowired private JmsOperations jmsOperations; public void sendMessage(final String message) { jmsOperations.send("biz1.queue", new MessageCreator() { public Message createMessage(Session session) throws JMSException { return session.createTextMessage(message); } }); } } </code>

我們可以看到，我們業務的實現中註入了一個JmsOperations對象，這個對象就是JmsTempate的實現。JmsOperations的send()方法有兩個參數，第一個是消息的destination，第二個便是具體的Message，在上面例子中message是通過一個匿名內部類MessageCreator的createMessage()方法構造的。

通過上面例子可以發現，通過JmsTempate，我們只需要關心發送消息即可，所有的連接和session的維護都由JmsTempate負責。

設置默認destination

大部分情況下，一個業務消息的destination是相同的，所以我們不必每次發送都填寫destination，我們可以在配置文件中對其進行配置：

?

1 2	<code><bean c:_-ref="connectionFactory" class="org.springframework.jms.core.JmsTemplate" id="jmsTemplate" p:defaultdestinationname="biz1.queue"> </bean></code>

在上面配置中我們默認destination值為biz1.queue，因為它只是聲明了一個名稱，並沒有說明是哪種類型的destination，所以，如果存在相同名稱的queue或topic，就會自動與之匹配，如果不存在，則會默認創建一個相同名稱的queue。如果我們想指定destination的類型，我們可以通過配置讓其依賴之前配置的destination bean即可：

?

1 2	<code><bean c:_-ref="connectionFactory" class="org.springframework.jms.core.JmsTemplate" id="jmsTemplate" p:defaultdestination-ref="biz1.Topic"> </bean></code>

當我們配置了默認destination，我們就可以在發送消息時省略第一個參數了：

?

1 2 3 4 5 6

<code>jmsOperations.send( new MessageCreator() { ... } ); </code>

其實上面的send()方法可以變得更簡單，我們可以利用消息轉換器。

使用消息轉換器發送消息

除了send()方法之外，JmsTemplate還提供了convertAndSend()方法。與send()方法需要依賴一個MessageCreator不同，convertAndSend()方法只需要傳入你想發送的消息即可。下面我們用convertAndSend()實現接口中的sendMessage()方法：

?

1 2 3 4

<code>public void sendMessage(final String message) { jmsOperations.convertAndSend(message); } </code>

convertAndSend()方法會自動把你發的消息轉換成Message，具體如何轉換的由org.springframework.messaging.converter.MessageConverter的實現來決定。我們先看一下MessageConverter接口：

?

1 2 3 4 5

<code>public interface MessageConverter { Object fromMessage(Message<!--?--> var1, Class<!--?--> var2); Message<!--?--> toMessage(Object var1, MessageHeaders var2); } </code>

我們可以看到這個接口中只有兩個方法而且很容易實現。其實大部分情況下我們不需要自己去實現這個接口，Spring已經為我們提供給了很多常用的實現：

默認情況下，當JmsTemplate的convertAndSend()方法使用的是SimpleMessageConverter。但是我們也可以通過配置把我們自定義的MessageConverter作為屬性註入到JmsTemplate中，比如我們有個一MessageConverter的實現bean：

?

1 2	<code><bean class="org.springframework.jms.support.converter.MappingJacksonMessageConverter" id="messageConverter"> </bean></code>

我們可以把上面這個bean註入到JmsTemplate中：

?

1 2	<code><bean c:_-ref="connectionFactory" class="org.springframework.jms.core.JmsTemplate" id="jmsTemplate" p:defaultdestinationname="spittle.alert.queue" p:messageconverter-ref="messageConverter"> </bean></code>

消費消息

對於消費來說，JmsTemplate使用起來比發送更簡單，只需要調用JmsOperations的receive()方法即可：

?

1 2 3 4 5 6 7 8 9 10 11 12 13 14

<code>public class ReceiveMessage { @Autowired private JmsOperations jmsOperations; public String receive() { try { ObjectMessage message = (ObjectMessage) jmsOperations.receive(); return (String) message.getObject(); } catch (JMSException e) { e.printStackTrace(); throw JmsUtils.convertJmsAccessException(e); } } } </code>

當調用 jmsOperations.receive()方法時，它會嘗試從broker獲取消息，若此時沒有消息，receive()方法會一直等待直到有消息產生。前面例子中，當我們發送消息的時候消息被封裝成的是ObjectMessage，所我們在獲取的時候可以再將其轉換回ObjectMessage。

這裏有一點需要註意，當調用message.getObject()方法時會拋出JMSException，這個異常是屬於JMS API的。JMSException是一個檢查異常，在JMS操作中會拋出各種各樣的JMSException，但是前面我們使用JmsTemplate時並沒有捕獲任何JMSException，是因為JmsTemplate內部已經將需要檢查的JMSException轉換成了非檢查的Spring自己的JmsException。在上面代碼中因為調用的是message.getObject()方法而不是JmsTemplate的方法，所以我們需要捕獲JMSException。但是按照Spring的設計理念，我們應該盡量減少檢查異常，所以在catch塊裏面我們又通過JmsUtils工具把JMSException轉換成了非檢查的JmsException。

同樣，就行消息的發送一樣，我們也可以使用JmsTemplate的receiveAndConvert()方法替換receive()方法：

?

1 2 3 4

<code>public String receive() { return (String)jmsOperations.receiveAndConvert(); } </code>

我們看到，因為使用的是JmsTemplate的方法，所以我們不需要再捕獲JMSException檢查異常。

不管使用msTemplate的receive()還是receiveAndConvert()方法消費消息，它們都是同步的。也就是說接收者在消息到達時需要等待。這樣看起來是不是有點奇怪？發送消息時是異步的，接收消息時卻是同步的。<喎?"/kf/ware/vc/" target="_blank" class="keylink">vcD4NCjxwPtXi0rK+zcrHzqrKssO0u+HT0M/Cw+a1xM/7z6LH/bavUE9KT7P2z9a1xNSt0vKjrM/Cw+bO0sPHvs2/tNK7z8LI57rOyrXP1tLssr21xL3TytXP+8+ioaM8L3A+DQo8aDMgaWQ9"23-創建消息驅動pojo">2.3 創建消息驅動POJO

我們上面已經知道，JmsTemplate的receive()方法是一個同步方法，在消息到達之前這個方法會掛起一直等待直到消息出現，如果這樣的話，我們的應用可能會出現一直等待消息而不能做其它事情的情況。為何不讓應用先去處理其它業務，當消息出現時再告知應用處理呢？

在EJB中，message driven bean（MDB）就可以實現異步的處理消息。Spring在這方面參考了EJB3對MDB的實現，不過在Spring中我們把它稱作消息驅動POJO，也就是message-driven POJO（MDP）。

2.3.1 創建一個消息監聽器

要想在消息出現時得到通知，那麽就需要一個監聽器監聽queue或者topic，之所以稱作消息驅動POJO，意識因為監聽器是消息驅動的，而是因為這個監聽器本身就是一個普通的POJO對象，不需要依賴任何接口：

?

1 2 3 4 5 6

<code>public class MyMessageHandler { public void handleMessage(String message){ //具體的實現 } } </code>

有了這個POJO對象，下面只需要做簡單的配置即可。

2.3.2 配置消息監聽器

賦予上面POJO接收消息能力的關鍵在於將其配置成一個Spring消息監聽器，Spring的jms命名空間提供了所有相關配置。

首先，我們現需要把上面的POJO對象聲明成一個bean：

?

1 2	<code><bean class="com.heaven.springexamples.jms.MyMessageHandler" id="myMessageHandler"> </bean></code>

其次，把MessageHandler變成一個消息驅動POJO，即把這個bean聲明成一個listener：

?

1 2 3 4

<code><jms:listener-container connection-factory="connectionFactory"> <jms:listener destination="biz1.queue" method="handleMessage" ref="myMessageHandler"> </jms:listener></jms:listener-container> </code>

通過上面配置，消息監聽容器裏面就多了一個消息監聽器。消息監聽容器是一個特殊的bean，它能夠監聽JMS的destination，監聽消息的到達。一旦消息到達，消息監聽容器會接受這個消息並將其發送給所有相關的listener。下面這幅圖展示了整個內部處理過程：

為了配置監聽容器和監聽者，我們用到了jms命名空間中的兩個元素。是父元素，是子元素。依賴一個connectionFactory，這樣它的各個就可以監聽消息了。用來定義具體接收消息的bean及方法。按照上面的配置，當消息到達queue時，MyMessageHandler的handleMessage方法便會被調用。

2.3.3 另一種方式，實現一個MessageListener接口

需要註意到是，我們的MessageHandler還可以實現一個MessageListener接口，這樣的話就不需要再單獨指定消息處理的方法了，MyMessageHandler的onMessage()方法會自動被調用。MessageListener接口定義如下：

?

1 2 3 4

<code>public interface MessageListener { void onMessage(Message var1); } </code>

我們寫一個簡單的實現類：

?

1 2 3 4 5 6

<code>public class MyMessageListener implements MessageListener{ public void onMessage(Message message) { //具體的實現 } } </code>

然後直接配置listener即可（不用再配置method方法屬性）：

?

1 2 3

<jms:listener-container connection-factory="connectionFactory"> <jms:listener destination="biz1.queue" ref="myMessageHandler"> </jms:listener></jms:listener-container>

Spring整合JMS(消息中間件)