Queue Affinity 和 LocalizedQueueConnectionFactory

發(fā)布者確認(rèn)和返回

記錄通道關(guān)閉事件

• 不記錄通道正常關(guān)閉事件 (200 OK).
• 如果通道是因?yàn)槭〉谋粍拥年?duì)列聲明關(guān)閉的，將記錄為debug級別.
• 如果通道關(guān)閉是因?yàn)?span style="color: #6d180b; font-family: Monaco, Consolas, Courier, 'Lucida Console', monospace; background-color: #f2f2f2;">basic.consume因專用消費(fèi)者條件而拒絕引起的,將被記錄為INFO級別.
• 所有其它的事件將記錄為ERROR級別.

運(yùn)行時緩存屬性

Table 3.1. CacheMode.CHANNEL的緩存屬性

channelCacheSize

localPort

idleChannelsTx

idleChannelsNotTx

idleChannelsTxHighWater

idleChannelsNotTxHighWater

CachingConnectionFactory 現(xiàn)在允許你訪問底層連接工廠,例如, 設(shè)置自定義client 屬性:

connectionFactory.getRabbitConnectionFactory().getClientProperties().put("foo", "bar");

當(dāng)在RabbitMQ管理界面中查看連接時，將會看到這些屬性．

3.1.4 AmqpTemplate

介紹

從１.3版本開始， 你可為RabbitTemplate 配置使用 RetryTemplate 來幫助處理broker連接的問題. 參考spring-retry 項(xiàng)目來了解全部信息;下面就是一個例子，它使用指數(shù)回退策略(exponential back off policy)和默認(rèn)的 SimpleRetryPolicy (向調(diào)用者拋出異常前，會做三次嘗試）.

使用XML命名空間:

<rabbit:template　id="template"　connection-factory="connectionFactory"　retry-template="retryTemplate"/>

<bean　id="retryTemplate"　class="org.springframework.retry.support.RetryTemplate">

<property　name="backOffPolicy">

<bean　class="org.springframework.retry.backoff.ExponentialBackOffPolicy">

<property　name="initialInterval"　value="500" />

<property　name="multiplier"　value="10.0" />

<property　name="maxInterval"value="10000" />

</bean>

</property>

</bean>

使用 @Configuration:

@Bean

public AmqpTemplate rabbitTemplate();
		RabbitTemplate template = new RabbitTemplate(connectionFactory());
		RetryTemplate retryTemplate = new RetryTemplate();
		ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
		backOffPolicy.setInitialInterval(500);
		backOffPolicy.setMultiplier(10.0);
		backOffPolicy.setMaxInterval(10000);
		retryTemplate.setBackOffPolicy(backOffPolicy);
		template.setRetryTemplate(retryTemplate);
		return template;
}

從1.4版本開始,除了retryTemplate 屬性外,RabbitTemplate 上也支持recoveryCallback 選項(xiàng). 它可用作RetryTemplate.execute(RetryCallback<T, E> retryCallback, RecoveryCallback<T>recoveryCallback)第二個參數(shù).

RecoveryCallback 會有一些限制，因?yàn)樵趓etry context只包含lastThrowable 字段.在更復(fù)雜的情況下，你應(yīng)該使用外部RetryTemplate,這樣你就可以通過上下文屬性傳遞更多信息給RecoveryCallback．

retryTemplate.execute(
    new RetryCallback<Object, Exception>() {

        @Override

　　public Object doWithRetry(RetryContext context) throws Exception {
            context.setAttribute("message", message);
            return rabbitTemplate.convertAndSend(exchange, routingKey, message);
        }
    }, new RecoveryCallback<Object>() {

        @Overridepublic Object recover(RetryContext context) throws Exception {
            Object message = context.getAttribute("message");
            Throwable t = context.getLastThrowable();
            // Do something with message

　　　return null;
        }
    });
}

在這種情況下，你不需要在RabbitTemplate中注入RetryTemplate.

發(fā)布者確認(rèn)和返回

Messaging 集成

從１.６版本開始，模板支持user-id-expression (當(dāng)使用Java配置時，為userIdExpression). 如果發(fā)送消息，user id屬性的值將在評估表達(dá)式后進(jìn)行設(shè)置．評價(jià)的根對象是要發(fā)送的消息。

例子:

<rabbit:template...user-id-expression="'guest'" />

<rabbit:template...user-id-expression="@myConnectionFactory.username" />

第一個示例是一個文本表達(dá)式；第二個例子將獲取上下文中連接工廠bean的username 屬性．

3.1.5 發(fā)送消息

當(dāng)發(fā)送消息時，可使用下面的任何一種方法：

void send(Message message) throws AmqpException;

void send(String routingKey, Message message) throws AmqpException;

void send(String exchange, String routingKey, Message message) throws AmqpException;

我們將使用上面列出的最后一個方法來討論，因?yàn)樗鼘?shí)際是最清晰的.它允許在運(yùn)行時提供一個AMQP Exchange 名稱和路由鍵(routing key).最后一個參數(shù)是負(fù)責(zé)初建創(chuàng)建Message實(shí)例的回調(diào).使用此方法來發(fā)送消息的示例如下:

amqpTemplate.send("marketData.topic", "quotes.nasdaq.FOO",
    new Message("12.34".getBytes(), someProperties));

如果你打算使用模板實(shí)例來多次(或多次)向同一個交換器發(fā)送消息時，"exchange" 可設(shè)置在模板自已身上.在這種情況中，可以使用上面列出的第二個方法. 下面的例子在功能上等價(jià)于前面那個：

amqpTemplate.setExchange("marketData.topic");
amqpTemplate.send("quotes.nasdaq.FOO", new Message("12.34".getBytes(), someProperties));

如果在模塊上設(shè)置"exchange"和"routingKey"屬性,那么方法就只接受Message 參數(shù):

amqpTemplate.setExchange("marketData.topic");
amqpTemplate.setRoutingKey("quotes.nasdaq.FOO");
amqpTemplate.send(new Message("12.34".getBytes(), someProperties));

關(guān)于交換器和路由鍵更好的想法是明確的參數(shù)將總是會覆蓋模板默認(rèn)值.事實(shí)上, 即使你不在模板上明確設(shè)置這些屬性, 總是有默認(rèn)值的地方. 在兩種情況中，默認(rèn)值是空字符串，這是合情合理的． 
就路由鍵而言，它并不總是首先需要的 (如． Fanout 交換器). 此外,綁定的交換器上的隊(duì)列可能會使用空字符串. 這些在模板的路由鍵中都是合法的.
就交換器名稱而言，空字符串也是常常使用的，因?yàn)锳MQP規(guī)范定義了無名稱的"默認(rèn)交換器".
由于所有隊(duì)列可使用它們的隊(duì)列名稱作為路由鍵自動綁定到默認(rèn)交換器上(它是Direct交換器e) ,上面的第二個方法可通過默認(rèn)的交換器將簡單的點(diǎn)對點(diǎn)消息傳遞到任何隊(duì)列.
只需要簡單的將隊(duì)列名稱作為路由鍵-在運(yùn)行時提供方法參數(shù)：

RabbitTemplate template = new RabbitTemplate(); // 使用默認(rèn)的無名交換器
template.send("queue.helloWorld", new Message("Hello World".getBytes(), someProperties));

或者，如果你喜歡創(chuàng)建一個模板用于主要或?qū)ｉT向一個隊(duì)列發(fā)送消息, 以下是完全合理的：

RabbitTemplate template = new RabbitTemplate(); // 使用默認(rèn)無名交換器
template.setRoutingKey("queue.helloWorld"); // 但我們總是向此隊(duì)列發(fā)送消息
template.send(new Message("Hello World".getBytes(), someProperties));

Message Builder API
從1.3版本開始,通過 MessageBuilder 和 MessagePropertiesBuilder提供了消息構(gòu)建API; 它們提供了更加方便地創(chuàng)建消息和消息屬性的方法:
Message message = MessageBuilder.withBody("foo".getBytes())
	.setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN)
	.setMessageId("123")
	.setHeader("bar", "baz")
	.build();
或
MessageProperties props = MessagePropertiesBuilder.newInstance()
	.setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN)
	.setMessageId("123")
	.setHeader("bar", "baz")
	.build();
Message message = MessageBuilder.withBody("foo".getBytes())
	.andProperties(props)
	.build();
每個MessageProperies上定義的屬性都可以被設(shè)置. 其它方法包括setHeader(String key, String value),removeHeader(String key), removeHeaders(), 和copyProperties(MessageProperties properties). 
每個屬性方法都有一個set*IfAbsent() 變種. 在默認(rèn)的初始值存在的情況下, 方法名為set*IfAbsentOrDefault().
提供了五個靜態(tài)方法來創(chuàng)建初始message builder:
public　static MessageBuilder withBody(byte[] body) 
public　static MessageBuilder withClonedBody(byte[] body) 
public　static MessageBuilder withBody(byte[] body, int from, int to) 
public　static MessageBuilder fromMessage(Message message) 
public　static MessageBuilder fromClonedMessage(Message message) 
builder創(chuàng)建的消息body是參數(shù)的直接引用.
builder創(chuàng)建的消息body是包含拷貝原字節(jié)數(shù)組的新數(shù)組.
build創(chuàng)建的消息body是包含原字節(jié)數(shù)組范圍的新數(shù)組.查看Arrays.copyOfRange() 來了解更多信息.
builder創(chuàng)建的消息body是原body參數(shù)的直接引用. 參數(shù)的屬性將拷貝到新MessageProperties對象中.
builer創(chuàng)建的消息body包含參數(shù)body的新數(shù)組.參數(shù)的屬性將拷貝到新的MessageProperties 對象中.
public static MessagePropertiesBuilder newInstance() 
public static MessagePropertiesBuilder fromProperties(MessageProperties properties) 
public static MessagePropertiesBuilder fromClonedProperties(MessageProperties properties) 
新消息屬性將使用默認(rèn)值進(jìn)行初始化
builder會使用提供的properties對象進(jìn)行初始化，build() 方法也會返回參數(shù)properties對象．
參數(shù)的屬性會拷貝到新的MessageProperties對象中.
在AmqpTemplate的RabbitTemplate 實(shí)現(xiàn)中, 每個send() 方法的重載版本都接受一個額外的CorrelationData對象.
當(dāng)啟用了發(fā)布者確認(rèn)時，此對象會在3.1.4, “AmqpTemplate”的回調(diào)中返回.這允許發(fā)送者使用確認(rèn)(ack或nack)來關(guān)聯(lián)發(fā)送的消息．
發(fā)布者返回
當(dāng)模板的mandatory 屬性為true時，返回消息將由 Section 3.1.4, “AmqpTemplate”描述的回調(diào)來返回.
從１.４版本開始，RabbitTemplate 支持 SpEL mandatoryExpression 屬性,它將對每個請求消息進(jìn)行評估，作為根評估對象來解析成布爾值. Bean引用,如"@myBean.isMandatory(#root)" 可用在此表達(dá)式中．
發(fā)布者返回內(nèi)部也可用于RabbitTemplate 的發(fā)送和接收操作中. 參考the section called “Reply Timeout” 來了解更多信息.
批量
從１.４.２版本開始，引入了BatchingRabbitTemplate.它是RabbitTemplate 的子類，覆蓋了send 方法，此方法可根據(jù)BatchingStrategy來批量發(fā)送消息; 只有當(dāng)一個批次完成時才會向RabbitMQ發(fā)送消息。
public interface BatchingStrategy {

	MessageBatch addToBatch(String exchange, String routingKey, Message message);

	Date nextRelease();

	Collection<MessageBatch> releaseBatches();

}

• batchSize - 發(fā)送前一個批次中消息的數(shù)量
• bufferLimit - 批量消息的最大大小;如果超過了此值，它會取代batchSize, 并導(dǎo)致要發(fā)送的部分批處理
• timeout - 當(dāng)沒有新的活動添加到消息批處理時之后，將發(fā)送部分批處理的時間(a time after which a partial batch will be sent when there is no new activity adding messages to the batch)

3.1.6 接收消息

介紹

Message 接收總是比發(fā)送稍顯復(fù)雜.有兩種方式來接收Message. 最簡單的選擇是在輪詢方法調(diào)用中一次只接收一個消息. 更復(fù)雜的更常見的方法是注冊一個偵聽器，按需異步的接收消息。
在下面兩個子章節(jié)中，我們將看到這兩種方法的示例.

AmqpTemplate 自身可用來輪詢消息接收.默認(rèn)情況下，如果沒有可用消息,將會立即返回 null;它是無阻塞的.
從1.5版本開始，你可以設(shè)置receiveTimeout,以毫秒為單位, receive方法會阻塞設(shè)定的時間來等待消息.小于０的值則意味著無限期阻塞 (或者至少要等到與broker的連接丟失).
1.6版本引入了receive 方法的變種，以允許在每個調(diào)用上都可設(shè)置超時時間.

警告

由于接收操作會為每個消息創(chuàng)建一個新的QueueingConsumer,這種技術(shù)并不適用于大容量環(huán)境，可考慮使用異步消費(fèi)者，或?qū)?/span>receiveTimeout 設(shè)為０來應(yīng)對這種情況．

這里有四個簡單可用的receive 方法.同發(fā)送方的交換器一樣, 有一種方法需要直接在模板本身上設(shè)置的默認(rèn)隊(duì)列屬性, 還有一種方法需要在運(yùn)行接受隊(duì)列參數(shù).
版本1.6 引入了接受timeoutMillis 的變種，基于每個請求重寫了receiveTimeout 方法.

Message receive() throws AmqpException;

Message receive(String queueName) throws AmqpException;

Message receive(long timeoutMillis) throws AmqpException;

Message receive(String queueName, long timeoutMillis) throws AmqpException;

與發(fā)送消息的情況類似， AmqpTemplate 有一些便利的方法來接收POJOs 而非Message 實(shí)例, 其實(shí)現(xiàn)可提供一種方法來定制MessageConverter 以用于創(chuàng)建返回的Object:

Object receiveAndConvert() throws AmqpException;

Object receiveAndConvert(String queueName) throws AmqpException;

Message receiveAndConvert(long timeoutMillis) throws AmqpException;

Message receiveAndConvert(String queueName, long timeoutMillis) throws AmqpException;

類似于sendAndReceive 方法,從1.3版本開始， AmqpTemplate 有多個便利的receiveAndReply 方法同步接收，處理，以及回應(yīng)消息:

<R, S> boolean receiveAndReply(ReceiveAndReplyCallback<R, S> callback)
	   throws AmqpException;

<R, S> boolean receiveAndReply(String queueName, ReceiveAndReplyCallback<R, S> callback)
 	throws AmqpException;

<R, S> boolean receiveAndReply(ReceiveAndReplyCallback<R, S> callback,
	String replyExchange, String replyRoutingKey) throws AmqpException;

<R, S> boolean receiveAndReply(String queueName, ReceiveAndReplyCallback<R, S> callback,
	String replyExchange, String replyRoutingKey) throws AmqpException;

<R, S> boolean receiveAndReply(ReceiveAndReplyCallback<R, S> callback,
 	ReplyToAddressCallback<S> replyToAddressCallback) throws AmqpException;

<R, S> boolean receiveAndReply(String queueName, ReceiveAndReplyCallback<R, S> callback,
			ReplyToAddressCallback<S> replyToAddressCallback) throws AmqpException;

AmqpTemplate 實(shí)現(xiàn)會負(fù)責(zé)receive 和 reply 階段.在大多數(shù)情況下，如果有必要，你只需要提供ReceiveAndReplyCallback 的實(shí)現(xiàn)來為收到的消息執(zhí)行某些業(yè)務(wù)邏輯或?yàn)槭盏降南?gòu)建回應(yīng)對象.
注意，ReceiveAndReplyCallback 可能返回null. 在這種情況下，將不會發(fā)送回應(yīng)，receiveAndReply 的工作類似于receive 方法. 這允許相同的隊(duì)列用于消息的混合物，其中一些可能不需要答復(fù)。

自動消息(請求和應(yīng)答)轉(zhuǎn)換只能適應(yīng)于提供的回調(diào)不是ReceiveAndReplyMessageCallback 實(shí)例的情況下- 它提供了一個原始的消息交換合同。

ReplyToAddressCallback 只在這種情況中有用，需要根據(jù)收到的信息通過自定義邏輯來決定replyTo 地址，并在ReceiveAndReplyCallback中進(jìn)行回應(yīng)的情況. 默認(rèn)情況下，請求消息中的 replyTo 信息用來路由回復(fù).

下面是一個基于POJO的接收和回復(fù)…​

boolean received =
        this.template.receiveAndReply(ROUTE, new ReceiveAndReplyCallback<Order, Invoice>() {

                public Invoice handle(Order order) {
                        return processOrder(order);
                }
        });
if (received) {
        log.info("We received an order!");
}

異步消費(fèi)者

對于異步消息接收, 會涉及到一個專用組件(不是AmqpTemplate).此組件可作為消息消費(fèi)回調(diào)的容器.
稍后，我們會講解這個容器和它的屬性,但首先讓我們來看一下回調(diào)，因?yàn)檫@里是你的應(yīng)用程序代碼與消息系統(tǒng)集成的地方. MessageListener 接口:

public　interface MessageListener {
    void onMessage(Message message);
}

如果出于任何理由，你的回調(diào)邏輯需要依賴于AMQP Channel實(shí)例,那么你可以使用ChannelAwareMessageListener. 它看起來是很相似的，但多了一個額外的參數(shù)：

public interface ChannelAwareMessageListener {
    void onMessage(Message message, Channel channel) throws Exception;
}

MessageListenerAdapter listener = new MessageListenerAdapter(somePojo);
    listener.setDefaultListenerMethod("myMethod");

你也可以繼承適配器，并實(shí)現(xiàn)getListenerMethodName()方法(基于消息來動態(tài)選擇不同的方法). 這個方法有兩個參數(shù)：originalMessage 和extractedMessage, 后者是轉(zhuǎn)換后的結(jié)果.默認(rèn)情況下，需要配置SimpleMessageConverter ;
參考the section called “SimpleMessageConverter” 來了解更多信息以及其它轉(zhuǎn)換器的信息.

從1.4.2開始，原始消息包含consumerQueue 和 consumerTag 屬性，這些屬性可用來確定消息是從那個隊(duì)列中收到的．

從1.5版本開始，你可以配置消費(fèi)者queue/tag到方法名稱的映射(map)以動態(tài)選擇要調(diào)用的方法.如果map中無條目，我們將退回到默認(rèn)監(jiān)聽器方法．

你已經(jīng)看過了消息監(jiān)聽回調(diào)上的各種各樣的選項(xiàng)，現(xiàn)在我們將注意力轉(zhuǎn)向容器. 基本上，容器處理主動(active)的職責(zé),這樣監(jiān)聽器回調(diào)可以保持被動(passive）. 容器是“生命周期”組件的一個例子。
它提供了啟動和停止的方法.當(dāng)配置容器時，你本質(zhì)上縮短了AMQP Queue和 MessageListener 實(shí)例之間的距離.你必須提供一個ConnectionFactory 的引用，隊(duì)列名稱或隊(duì)列實(shí)例．
下面是使用默認(rèn)實(shí)現(xiàn)SimpleMessageListenerContainer 的最基礎(chǔ)的例子:

SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setConnectionFactory(rabbitConnectionFactory);
container.setQueueNames("some.queue");
container.setMessageListener(new MessageListenerAdapter(somePojo));

作為一個主動組件, 最常見的是使用bean定義來創(chuàng)建監(jiān)聽器容器，這樣它就可以簡單地運(yùn)行于后臺.這可以通過XML來完成:

<rabbit:listener-container　connection-factory="rabbitConnectionFactory">

<rabbit:listener　queues="some.queue"　ref="somePojo"　method="handle"/>

</rabbit:listener-container>

或者你可以@Configuration 風(fēng)格:

@Configuration

public　class ExampleAmqpConfiguration {

    @Bean

public SimpleMessageListenerContainer messageListenerContainer() {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(rabbitConnectionFactory());
        container.setQueueName("some.queue");
        container.setMessageListener(exampleListener());
        return container;
    }

    @Bean

public ConnectionFactory rabbitConnectionFactory() {
        CachingConnectionFactory connectionFactory =
            new CachingConnectionFactory("localhost");
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        return connectionFactory;
    }

    @Bean

public MessageListener exampleListener() {
        returnnew MessageListener() {
            publicvoid onMessage(Message message) {
                System.out.println("received: " + message);
            }
        };
    }
}

從RabbitMQ Version 3.2開始, broker支持消費(fèi)者優(yōu)先級了(參考 Using Consumer Priorities with RabbitMQ). 

這可以通過在消費(fèi)者設(shè)置x-priority 參數(shù)來啟用． 

SimpleMessageListenerContainer 現(xiàn)在支持設(shè)置消費(fèi)者參數(shù):

container.setConsumerArguments(Collections.
<String, Object> singletonMap("x-priority", Integer.valueOf(10)));

為了方便,命名空間在listener元素上提供了priority 屬性:

<rabbit:listener-container　connection-factory="rabbitConnectionFactory">

<rabbit:listener　queues="some.queue"　ref="somePojo"　method="handle"　priority="10" />

</rabbit:listener-container>

從１.３版本開始，容器監(jiān)聽的隊(duì)列可在運(yùn)行時進(jìn)行修改，參考 Section 3.1.18, “Listener Container Queues”.

當(dāng)容器配置為監(jiān)聽auto-delete 隊(duì)列或隊(duì)列有x-expires 選項(xiàng)或者broker配置了Time-To-Live 策略,隊(duì)列將在容器停止時(最后的消費(fèi)者退出時)由broker進(jìn)行刪除.
在１.３版本之前，容器會因隊(duì)列缺失而不能重啟; 當(dāng)連接關(guān)閉／打開時，RabbitAdmin 只能自動重新聲明隊(duì)列.

從1.3版本開始, 在啟動時，容器會使用RabbitAdmin 來重新聲明缺失的隊(duì)列．

您也可以使用條件聲明(the section called “Conditional Declaration”) 與auto-startup="false" 來管理隊(duì)列的延遲聲明，直到容器啟動.

<rabbit:queue　id="otherAnon"　declared-by="containerAdmin" />

<rabbit:direct-exchange　name="otherExchange"　auto-delete="true"　declared-by="containerAdmin">

<rabbit:bindings>

<rabbit:binding　queue="otherAnon"　key="otherAnon" />

</rabbit:bindings>

</rabbit:direct-exchange>

<rabbit:listener-container　id="container2"　auto-startup="false">

<rabbit:listener　id="listener2"　ref="foo"queues="otherAnon"　admin="containerAdmin" />

</rabbit:listener-container>

<rabbit:admin　id="containerAdmin"　connection-factory="rabbitConnectionFactory"　auto-startup="false" />

在這種情況下，隊(duì)列和交換器是由 containerAdmin 來聲明的，auto-startup="false" 因此在上下文初始化期間不會聲明元素.同樣，出于同樣原因，容器也不會啟動.當(dāng)容器隨后啟動時，它會使用containerAdmin引用來聲明元素.

批量消息

消費(fèi)者失敗事件

• container - 消費(fèi)者經(jīng)歷問題的監(jiān)聽容器．
• reason - 失敗的文本原因。
• fatal - 一個表示失敗是否是致命的boolean值;對于非致命異常,容器會根據(jù)retryInterval值嘗試重新啟動消費(fèi)者.
• throwable -捕捉到的Throwable.