介紹
Spring AMQP 由少數(shù)幾個(gè)模塊組成, 每個(gè)都以JAR的形式來(lái)表現(xiàn).這些模塊是: spring-amqp和spring-rabbit. spring-amqp模塊包含org.springframework.amqp.core 包.
在那個(gè)包中,你會(huì)找到表示AMQP核心模塊的類(lèi). 我們的目的是提供通用的抽象，不依賴(lài)于任何特定中間件的實(shí)現(xiàn)或AMQP客戶(hù)端庫(kù)。
最終用戶(hù)代碼將更具有移植性，以便跨供應(yīng)商實(shí)現(xiàn)，因?yàn)樗鼈兛梢詫?duì)抽象層開(kāi)發(fā)。這些抽象是使用broker的特定模塊實(shí)現(xiàn)的，如spring-rabbit。
目前只有一個(gè)RabbitMQ實(shí)現(xiàn)；而對(duì)于抽象的驗(yàn)證，除了RabbitMQ外,也已經(jīng)在.Net平臺(tái)上使用Apache Qpid得到了驗(yàn)證。
由于AMQP原則上工作于協(xié)議層次，RabbitMQ客戶(hù)端可以在任何支持相同的協(xié)議版本的broker中使用，但目前我們沒(méi)有測(cè)試其它任何broker。
這里的概述假設(shè)你已經(jīng)熟悉了AMQP規(guī)范.如果你還沒(méi)有，那么你可以查看第5章,其它資源中列舉的資源．
Message
AMQP 0-8 和0-9-1 規(guī)范沒(méi)有定義一個(gè)消息類(lèi)或接口.相反，當(dāng)執(zhí)行basicPublish()這樣的操作的時(shí)候, 內(nèi)容是以字節(jié)數(shù)組參數(shù)進(jìn)行傳遞的，其它額外屬性也是以單獨(dú)參數(shù)進(jìn)行傳遞的. Spring AMQP定義了一個(gè) Message 類(lèi)來(lái)作為AMQP領(lǐng)域模型表示的一部分.Message類(lèi)的目的是在單個(gè)實(shí)例中簡(jiǎn)化封裝消息體(body)和屬性(header),這樣API就可以變得更簡(jiǎn)單. Message類(lèi)的定義相當(dāng)簡(jiǎn)單．
public class Message {
private final MessageProperties messageProperties;
private final byte[] body;
public Message(byte[] body, MessageProperties messageProperties) {
this.body = body;
this.messageProperties = messageProperties;
}
public byte[] getBody() {
returnthis.body;
}
public MessageProperties getMessageProperties() {
returnthis.messageProperties;
}
}MessageProperties 接口定義了多個(gè)共同屬性，如messageId, timestamp, contentType 等等. 那些屬性可以通過(guò)調(diào)用setHeader(String key, Object value) 方法使用用戶(hù)定義頭(user-defined headers)來(lái)擴(kuò)展.
Exchange
Exchange 接口代表的是AMQP Exchange,它是生產(chǎn)者發(fā)送消息的地方.broker虛擬主機(jī)中的交換器名稱(chēng)都是唯一的，同時(shí)還有少量的其它屬性:
public interface Exchange {
String getName();
String getExchangeType();
boolean isDurable();
boolean isAutoDelete();
Map<String, Object> getArguments();
}正如你所看到的, Exchange還有一個(gè)type (它是在ExchangeTypes中定義的常量). 基本類(lèi)型是: Direct, Topic, Fanout,和Headers.
在核心包中，你可以找到每種類(lèi)型的Exchange 接口實(shí)現(xiàn).這些交換器類(lèi)型會(huì)在處理隊(duì)列綁定時(shí)，行為有所不同．
例如,Direct交換器允許隊(duì)列以固定路由鍵進(jìn)行綁定(通常是隊(duì)列的名稱(chēng)）.
Topic交換器支持使用路由正則表達(dá)式(*通配符明確匹配一個(gè)，而#通配符可匹配０個(gè)或多個(gè)）.
Fanout交換器會(huì)把消息發(fā)布到所有綁定到它上面的隊(duì)列而不考慮任何路由鍵.
關(guān)于交換器類(lèi)型的更多信息，查看Chapter 5, Other Resources.
AMQP規(guī)范還要求任何broker必須提供一個(gè)默認(rèn)的無(wú)名字的(空字符串)Direct交換器.所有聲明的隊(duì)列都可以用它們的名字作為路由鍵綁定到默認(rèn)交換器中. 在Section 3.1.4, “AmqpTemplate”你會(huì)了解到更多關(guān)于在Sring AMQP中使用默認(rèn)交換器的使用情況.
Queue
Queue 代表的是消費(fèi)者接收消息的組件. 像各種各樣的 Exchange 類(lèi)，我們的實(shí)現(xiàn)目標(biāo)是作為核心AMQP類(lèi)型的抽象表示.
public class Queue {
private final String name;
private volatile boolean durable;
private volatile boolean exclusive;
private volatile boolean autoDelete;
private volatile Map<String, Object> arguments;
/**
* The queue is durable, non-exclusive and non auto-delete.
*
* @param name the name of the queue.
*/ public Queue(String name) {
this(name, true, false, false);
}
// Getters and Setters omitted for brevity
}注意，構(gòu)造器需要接受隊(duì)列名稱(chēng)作為參數(shù)．根據(jù)實(shí)現(xiàn), admin template可能會(huì)提供生成獨(dú)特隊(duì)列名稱(chēng)的方法.這些隊(duì)列作為回復(fù)地址或用于臨時(shí)情景是非常有用的.
基于這種原因,自動(dòng)生成隊(duì)列的exclusive和 autoDelete 屬性都應(yīng)該設(shè)置為true.
參考 Section 3.1.10, “Configuring the broker” 來(lái)了解關(guān)于使用命名空間來(lái)聲明隊(duì)列，包括隊(duì)列參數(shù)的詳細(xì)情況.
Binding
生產(chǎn)者發(fā)送消息到Exchange，而消費(fèi)者將從Queue中獲取消息,連接Queues與Exchanges之間的綁定對(duì)于通過(guò)消息來(lái)連接生產(chǎn)者和消費(fèi)者是非常關(guān)鍵的.
在Spring AMQP中,我們定義了一個(gè) Binding 類(lèi)來(lái)表示這些連接. 讓我們重新回顧一下綁定隊(duì)列和交換器的操作.
你可以使用固定的路由鍵來(lái)綁定 Queue 到 DirectExchange上.
new Binding(someQueue, someDirectExchange, "foo.bar")
你可以使用路由正則表達(dá)式來(lái)綁定Queue到TopicExchange上.
new Binding(someQueue, someTopicExchange, "foo.*")
你可以不使用路由鍵來(lái)綁定Queue到FanoutExchange上.
new Binding(someQueue, someFanoutExchange)
我們還提供了BindingBuilder來(lái)方便操作.
Binding b = BindingBuilder.bind(someQueue).to(someTopicExchange).with("foo.*");上面展示的BindingBuilder 類(lèi)很清晰,但如果為bind()方法使用靜態(tài)導(dǎo)入，這種形式將工作得更好.
本身來(lái)說(shuō),Binding類(lèi)的實(shí)例只能一個(gè)connection中持有數(shù)據(jù).換句話(huà)說(shuō),它不是一個(gè)活力(active)組件.
但正如在后面Section 3.1.10, “Configuring the broker”看到的, Binding實(shí)例可由AmqpAdmin 類(lèi)來(lái)觸發(fā)broker上的綁定操作.
同樣，在同一個(gè)章節(jié)中，你還會(huì)看到Binding實(shí)例可在@Configuration類(lèi)中使用Spring @Bean風(fēng)格來(lái)定義.
還有方便的基類(lèi)來(lái)簡(jiǎn)化生成AMQP相關(guān)bean定義和識(shí)別隊(duì)列，交換器，綁定的方法，這樣當(dāng)AMQP broker運(yùn)行程序啟動(dòng)時(shí)，就可以得到聲明.
AmqpTemplate 也在核心包中定義．作為涉及AMQP消息的主要組件, 會(huì)在它自己的章節(jié)中進(jìn)行詳細(xì)介紹(參考Section 3.1.4, “AmqpTemplate”).
3.1.2 連接和資源管理
介紹
雖然我們?cè)谇懊嬲鹿?jié)中描述的AMQP模型是通用的，適用于所有實(shí)現(xiàn),但當(dāng)我們說(shuō)到資源管理時(shí)，其細(xì)節(jié)是針對(duì)特定broker實(shí)現(xiàn)的.因此，在這個(gè)章節(jié)中,我們只關(guān)注我們的"spring-rabbit"模塊，因?yàn)榈浆F(xiàn)在為止，RabbitMQ是唯一支持的實(shí)現(xiàn).
RabbitMQ broker中用于管理連接的中心組件是ConnectionFactory 接口. ConnectionFactory實(shí)現(xiàn)的責(zé)任是提供一個(gè)org.springframework.amqp.rabbit.connection.Connection 的實(shí)例，它包裝了com.rabbitmq.client.Connection.
我們提供的唯一具體實(shí)現(xiàn)提CachingConnectionFactory,默認(rèn)情況下，會(huì)建立應(yīng)用程序可共享的單個(gè)連接代理.連接共享是可行的，因?yàn)樵贏MQP處理消息的工作單位實(shí)際是 "channel" (在某些方面，這類(lèi)似于JMS中Connection 和 Sessionin的關(guān)系).
你可以想象，連接實(shí)例提供了一個(gè)createChannel方法。CachingConnectionFactory 實(shí)現(xiàn)支持這些channels的緩存,它會(huì)基于它們是否是事務(wù)的來(lái)單獨(dú)維護(hù)其緩存．
當(dāng)創(chuàng)建CachingConnectionFactory的實(shí)例時(shí), hostname 可通過(guò)構(gòu)造器來(lái)提供，username 和password 屬性也可以提供.如果你想配置channel緩存的大小(默認(rèn)是25),你可以調(diào)用setChannelCacheSize()方法.
從1.3版本開(kāi)始,CachingConnectionFactory 也可以同channel一樣，配置緩存連接.在這種情況下每次調(diào)用createConnection() 都會(huì)創(chuàng)建一個(gè)新連接(或者從緩存中獲取空閑的連接).
關(guān)閉連接會(huì)將其返回到緩存中(如果還沒(méi)有達(dá)到緩存大小的話(huà)).在這些連接上創(chuàng)建的Channels同樣也會(huì)被緩存. 單獨(dú)連接的使用在某些環(huán)境中是有用的，如從HA 集群中消費(fèi), 連接負(fù)載均衡器,連接不同的集群成員.設(shè)置cacheMode 為 CacheMode.CONNECTION.
這不會(huì)限制連接的數(shù)目，它用于指定允許空閑打開(kāi)連接的數(shù)目.
從１.5.5版本開(kāi)始,提供了一個(gè)新屬性connectionLimit.當(dāng)設(shè)置了此屬性時(shí)，它會(huì)限制連接的總數(shù)目，當(dāng)達(dá)到限制值時(shí)，將channelCheckoutTimeLimit 來(lái)等待空閑連接.如果時(shí)間超時(shí)了，將拋出AmqpTimeoutException.
重要
當(dāng)緩存模式是CONNECTION時(shí), 隊(duì)列的自動(dòng)聲明等等 (參考 the section called “Automatic Declaration of Exchanges, Queues and Bindings”) 將不再支持.
此外，在寫(xiě)作的時(shí)候，rabbitmq-client 包默認(rèn)為每個(gè)連接(５個(gè)線(xiàn)程)創(chuàng)建了一個(gè)固定的線(xiàn)程池. 當(dāng)需要使用大量連接時(shí)，你應(yīng)該考慮在CachingConnectionFactory定制一個(gè)executor. 然后，同一個(gè)executor會(huì)用于所有連接，其線(xiàn)程也是共享的．
executor的線(xiàn)程池是沒(méi)有界限的或按預(yù)期使用率來(lái)設(shè)置(通常, 一個(gè)連接至少應(yīng)該有一個(gè)線(xiàn)程）.如果在每個(gè)連接上創(chuàng)建了多個(gè)channels，那么池的大小會(huì)影響并發(fā)性,因此一個(gè)可變的線(xiàn)程池executor應(yīng)該是最合適的.
理解緩存大小不是限制是很重要的, 它僅僅是可以緩存的通道數(shù)量.當(dāng)說(shuō)緩存大小是１０時(shí)，在實(shí)際使用中，其實(shí)可以是任何數(shù)目的通道． 如果超過(guò)10個(gè)通道被使用，他們都返回到高速緩存，10個(gè)將在高速緩存中，其余的將物理關(guān)閉。
從1.6版本開(kāi)始,默認(rèn)通道緩存大小從１增加到了25. 在高容量,多線(xiàn)程環(huán)境中，較小的緩存意味著通道的創(chuàng)建和關(guān)閉將以很高的速率運(yùn)行.加大默認(rèn)緩存大小可避免這種開(kāi)銷(xiāo).
你可以監(jiān)控通道的使用情況(通過(guò)RabbitMQ Admin UI） ，如果看到有很多通道在創(chuàng)建和關(guān)閉，你可以增大緩存大小.緩存只會(huì)增長(zhǎng)按需（以適應(yīng)應(yīng)用程序的并發(fā)性要求），所以這個(gè)更改不會(huì)影響現(xiàn)有的低容量應(yīng)用程序。
從1.4.2版本開(kāi)始,CachingConnectionFactory 有一個(gè)channelCheckoutTimeout屬性. 當(dāng)此屬性的值大于０時(shí)，channelCacheSize 會(huì)變成連接上創(chuàng)建通道數(shù)目的限制.
如果達(dá)到了限制,調(diào)用線(xiàn)程將會(huì)阻塞，直到某個(gè)通道可用或者超時(shí), 在后者的情況中，將拋出AmqpTimeoutException 異常.
在框架(如．RabbitTemplate)中使用的通道將會(huì)可靠地返回到緩存中.如果在框架外創(chuàng)建了通道 (如.直接訪(fǎng)問(wèn)connection(s)并調(diào)用createChannel()),你必須可靠地返回它們(通過(guò)關(guān)閉),也許需要在 finally 塊中以防止耗盡通道.
CachingConnectionFactory connectionFactory = new CachingConnectionFactory("somehost");
connectionFactory.setUsername("guest");
connectionFactory.setPassword("guest");
Connection connection = connectionFactory.createConnection();當(dāng)使用XML時(shí),配置看起來(lái)像下面這樣:
<bean id="connectionFactory"
class="org.springframework.amqp.rabbit.connection.CachingConnectionFactory">
<constructor-arg value="somehost"/>
<property name="username"value="guest"/>
<property name="password" value="guest"/>
</bean>
這里也有一個(gè) SingleConnectionFactory 實(shí)現(xiàn)，它只能用于框架的單元測(cè)試代碼中.它比CachingConnectionFactory 簡(jiǎn)單，因?yàn)樗粫?huì)緩存通道,由于其缺乏性能和韌性，它不打算用于簡(jiǎn)單的測(cè)試以外的實(shí)際使用.
如果基于某些原因，你需要自己來(lái)實(shí)現(xiàn)ConnectionFactory ，AbstractConnectionFactory 基類(lèi)提供了一個(gè)非常好的起點(diǎn)．
ConnectionFactory 可使用rabbit命名空間來(lái)快速方便的建立：
<rabbit:connection-factory id="connectionFactory"/>
在多數(shù)情況下，這是很好的，因?yàn)榭蚣軙?huì)為你選擇最好的默認(rèn)值.創(chuàng)建的實(shí)例會(huì)是CachingConnectionFactory.要記住，默認(rèn)的緩存大小是25.如果你想緩存更多的通道，你可以設(shè)置channelCacheSize 屬性值.在XML中，它看起來(lái)像下面這樣:
<bean id="connectionFactory"
class="org.springframework.amqp.rabbit.connection.CachingConnectionFactory">
<constructor-arg value="somehost"/>
<property name="username" value="guest"/>
<property name="password" value="guest"/>
<property name="channelCacheSize" value="50"/>
</bean>
在命名空間中，你也可以添加channel-cache-size 屬性:
<rabbit:connection-factory id="connectionFactory" channel-cache-size="50"/>
默認(rèn)的緩存模式是CHANNEL, 但你可以使用緩存連接來(lái)替換;在這種情況下，我們會(huì)使用connection-cache-size:
<rabbit:connection-factory id="connectionFactory" cache-mode="CONNECTION" connection-cache-size="25"/>
Host 和 port 屬性也可以在命名空間中提供：
<rabbit:connection-factory id="connectionFactory" host="somehost" port="5672"/>
此外，如果運(yùn)行集群環(huán)境中，使用addresses屬性.
<rabbit:connection-factory id="connectionFactory" addresses="host1:5672,host2:5672"/>
下面是一個(gè)自定義的線(xiàn)程工廠(chǎng)，其前輟線(xiàn)程名稱(chēng)為rabbitmq-.
<rabbit:connection-factory id="multiHost" virtual-host="/bar" addresses="host1:1234,host2,host3:4567" thread-factory="tf" channel-cache-size="10" username="user" password="password" />
<bean id="tf" class="org.springframework.scheduling.concurrent.CustomizableThreadFactory">
<constructor-arg value="rabbitmq-" />
</bean>
配置底層客戶(hù)端連接工廠(chǎng)
CachingConnectionFactory 使用的是 Rabbit client ConnectionFactory的實(shí)例; 當(dāng)在CachingConnectionFactory設(shè)置等價(jià)屬性時(shí)，許多屬性(host, port, userName, password, requestedHeartBeat, connectionTimeout) 來(lái)傳遞.
要設(shè)置其它屬性(例如clientProperties),可定義一個(gè)rabbit factory 的實(shí)例，并使用CachingConnectionFactory的適當(dāng)構(gòu)造器來(lái)提供引用.
當(dāng)使用上面提到的命名空間時(shí)，要在connection-factory屬性中提供一個(gè)工廠(chǎng)的引用來(lái)配置. 為方便起見(jiàn)，提供了一個(gè)工廠(chǎng)，以協(xié)助在一個(gè)Spring應(yīng)用程序上下文中配置連接工廠(chǎng)，在下一節(jié)討論。
<rabbit:connection-factory id="connectionFactory" connection-factory="rabbitConnectionFactory"/>
RabbitConnectionFactoryBean 和配置SSL
從1.4版本開(kāi)始, 提供了一個(gè)便利的RabbitConnectionFactoryBean 類(lèi)通過(guò)依賴(lài)注入來(lái)配置底層客戶(hù)端連接工廠(chǎng)的SSL屬性.其它設(shè)置簡(jiǎn)單地委派給底層工廠(chǎng)．以前你必須以編程方式配置SSL選項(xiàng)。
<rabbit:connection-factory id="rabbitConnectionFactory"connection-factory="clientConnectionFactory" host="${host}" port="${port}" virtual-host="${vhost}" username="${username}" password="${password}" /><bean id="clientConnectionFactory" class="org.springframework.xd.dirt.integration.rabbit.RabbitConnectionFactoryBean">
<property name="useSSL" value="true" />
<property name="sslPropertiesLocation" value="file:/secrets/rabbitSSL.properties"/>
</bean>
參考 RabbitMQ Documentation 來(lái)了解關(guān)于配置SSL的更多信息. 省略的keyStore 和 trustStore 配置將在無(wú)證書(shū)驗(yàn)證的情況下，通過(guò)SSL來(lái)連接. Key和trust store 配置可以按如下提供：
sslPropertiesLocation 屬性是一個(gè)Spring Resource ，它指向一個(gè)包含下面key的屬性文件：
keyStore=file:/secret/keycert.p12
trustStore=file:/secret/trustStore
keyStore.passPhrase=secret
trustStore.passPhrase=secret
keyStore 的 truststore 是指向store的 Spring Resources .通常情況下，這個(gè)屬性文件在操作系統(tǒng)之下安全的，應(yīng)用程序只能讀取訪(fǎng)問(wèn).
從Spring AMQP １.５版本開(kāi)始，這些屬性可直接在工廠(chǎng)bean上設(shè)置.如果同時(shí)提供了discrete和 sslPropertiesLocation 屬性, 后者屬性值會(huì)覆蓋discrete值.
路由連接工廠(chǎng)
從1.3版本開(kāi)始,引入了AbstractRoutingConnectionFactory.這提供了一種機(jī)制來(lái)配置多個(gè)ConnectionFactories的映射，并通過(guò)在運(yùn)行時(shí)使用lookupKey來(lái)決定目標(biāo)ConnectionFactory.
通常,實(shí)現(xiàn)會(huì)檢查線(xiàn)程綁定上下文. 為了方便, Spring AMQP提供了SimpleRoutingConnectionFactory, 它會(huì)從SimpleResourceHolder中獲取當(dāng)前線(xiàn)程綁定的lookupKey:
<bean id="connectionFactory" class="org.springframework.amqp.rabbit.connection.SimpleRoutingConnectionFactory">
<propertyname="targetConnectionFactories">
<map>
<entry key="#{connectionFactory1.virtualHost}" ref="connectionFactory1"/> <entry key="#{connectionFactory2.virtualHost}" ref="connectionFactory2"/></map>
</property>
</bean>
<rabbit:template id="template" connection-factory="connectionFactory" />
public class MyService {
@Autowired private RabbitTemplate rabbitTemplate;
public void service(String vHost, String payload) {
SimpleResourceHolder.bind(rabbitTemplate.getConnectionFactory(), vHost);
rabbitTemplate.convertAndSend(payload);
SimpleResourceHolder.unbind(rabbitTemplate.getConnectionFactory());
}
}在使用資源后，對(duì)其進(jìn)行解綁是很重要的.更多信息參考AbstractRoutingConnectionFactory的JavaDocs.
從1.4版本開(kāi)始, RabbitTemplate 支持SpEL sendConnectionFactorySelectorExpression 和receiveConnectionFactorySelectorExpression 屬性,
它會(huì)在每個(gè)AMQP 協(xié)議交互操作(send, sendAndReceive, receiveor receiveAndReply)進(jìn)行評(píng)估, 為提供的AbstractRoutingConnectionFactory類(lèi)解析lookupKey值.
Bean 引用,如"@vHostResolver.getVHost(#root)" 可用于表達(dá)式中.對(duì)于send 操作, 要發(fā)送的消息是根評(píng)估對(duì)象;對(duì)于receive操作, queueName 是根評(píng)估對(duì)象.
路由算法為:如果selector 表達(dá)式為null,或等價(jià)于null,或提供的ConnectionFactory 不是AbstractRoutingConnectionFactory的實(shí)例,根據(jù)提供的ConnectionFactory 實(shí)現(xiàn)，
所有的工作都按之前的進(jìn)行.同樣的結(jié)果也會(huì)發(fā)生：如果評(píng)估結(jié)果不為null,但對(duì)于lookupKey 無(wú)目標(biāo)ConnectionFactory，且 the AbstractRoutingConnectionFactory 使用lenientFallback = true進(jìn)行了配置.
當(dāng)然，在AbstractRoutingConnectionFactory 的情況下，它會(huì)基于determineCurrentLookupKey()的路由實(shí)現(xiàn)來(lái)進(jìn)行回退. 但,如果lenientFallback = false, 將會(huì)拋出 IllegalStateException 異常.
Namespace 在<rabbit:template>組件中也支持send-connection-factory-selector-expression 和receive-connection-factory-selector-expression屬性．
也是從1.4版本開(kāi)始, 你可以在SimpleMessageListenerContainer配置路由連接工廠(chǎng). 在那種情況下，隊(duì)列名稱(chēng)的列表將作為lookup key.例如,如果你在容器中配置setQueueNames("foo", "bar"),lookup key將是"[foo,bar]" (無(wú)空格).