Queue Affinity 和 LocalizedQueueConnectionFactory

发布者确认和返回

记录通道关闭事件

• 不记录通道正常关闭事件 (200 OK).
• 如果通道是因为失败的被动的队列声明关闭的，将记录为debug级别.
• 如果通道关闭是因为basic.consume因专用消费者条件而拒绝引起的,将被记录为INFO级别.
• 所有其它的事件将记录为ERROR级别.

运行时缓存属性

Table 3.1. CacheMode.CHANNEL的缓存属性

channelCacheSize

localPort

idleChannelsTx

idleChannelsNotTx

idleChannelsTxHighWater

idleChannelsNotTxHighWater

CachingConnectionFactory 现在允许你访问底层连接工厂,例如, 设置自定义client 属性:

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

当在RabbitMQ管理界面中查看连接时，将会看到这些属性．

3.1.4 AmqpTemplate

介绍

从１.3版本开始， 你可为RabbitTemplate 配置使用 RetryTemplate 来帮助处理broker连接的问题. 参考spring-retry 项目来了解全部信息;下面就是一个例子，它使用指数回退策略(exponential back off policy)和默认的 SimpleRetryPolicy (向调用者抛出异常前，会做三次尝试）.

使用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 选项. 它可用作RetryTemplate.execute(RetryCallback<T, E> retryCallback, RecoveryCallback<T>recoveryCallback)第二个参数.

RecoveryCallback 会有一些限制，因为在retry context只包含lastThrowable 字段.在更复杂的情况下，你应该使用外部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.

发布者确认和返回

Messaging 集成

从１.６版本开始，模板支持user-id-expression (当使用Java配置时，为userIdExpression). 如果发送消息，user id属性的值将在评估表达式后进行设置．评价的根对象是要发送的消息。

例子:

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

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

第一个示例是一个文本表达式；第二个例子将获取上下文中连接工厂bean的username 属性．

3.1.5 发送消息

当发送消息时，可使用下面的任何一种方法：

void send(Message message) throws AmqpException;

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

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

我们将使用上面列出的最后一个方法来讨论，因为它实际是最清晰的.它允许在运行时提供一个AMQP Exchange 名称和路由键(routing key).最后一个参数是负责初建创建Message实例的回调.使用此方法来发送消息的示例如下:

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

如果你打算使用模板实例来多次(或多次)向同一个交换器发送消息时，"exchange" 可设置在模板自已身上.在这种情况中，可以使用上面列出的第二个方法. 下面的例子在功能上等价于前面那个：

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

如果在模块上设置"exchange"和"routingKey"属性,那么方法就只接受Message 参数:

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

关于交换器和路由键更好的想法是明确的参数将总是会覆盖模板默认值.事实上, 即使你不在模板上明确设置这些属性, 总是有默认值的地方. 在两种情况中，默认值是空字符串，这是合情合理的． 
就路由键而言，它并不总是首先需要的 (如． Fanout 交换器). 此外,绑定的交换器上的队列可能会使用空字符串. 这些在模板的路由键中都是合法的.
就交换器名称而言，空字符串也是常常使用的，因为AMQP规范定义了无名称的"默认交换器".
由于所有队列可使用它们的队列名称作为路由键自动绑定到默认交换器上(它是Direct交换器e) ,上面的第二个方法可通过默认的交换器将简单的点对点消息传递到任何队列.
只需要简单的将队列名称作为路由键-在运行时提供方法参数：

RabbitTemplate template = new RabbitTemplate(); // 使用默认的无名交换器
template.send("queue.helloWorld", new Message("Hello World".getBytes(), someProperties));

或者，如果你喜欢创建一个模板用于主要或专门向一个队列发送消息, 以下是完全合理的：

RabbitTemplate template = new RabbitTemplate(); // 使用默认无名交换器
template.setRoutingKey("queue.helloWorld"); // 但我们总是向此队列发送消息
template.send(new Message("Hello World".getBytes(), someProperties));

Message Builder API
从1.3版本开始,通过 MessageBuilder 和 MessagePropertiesBuilder提供了消息构建API; 它们提供了更加方便地创建消息和消息属性的方法:
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上定义的属性都可以被设置. 其它方法包括setHeader(String key, String value),removeHeader(String key), removeHeaders(), 和copyProperties(MessageProperties properties). 
每个属性方法都有一个set*IfAbsent() 变种. 在默认的初始值存在的情况下, 方法名为set*IfAbsentOrDefault().
提供了五个静态方法来创建初始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创建的消息body是参数的直接引用.
builder创建的消息body是包含拷贝原字节数组的新数组.
build创建的消息body是包含原字节数组范围的新数组.查看Arrays.copyOfRange() 来了解更多信息.
builder创建的消息body是原body参数的直接引用. 参数的属性将拷贝到新MessageProperties对象中.
builer创建的消息body包含参数body的新数组.参数的属性将拷贝到新的MessageProperties 对象中.
public static MessagePropertiesBuilder newInstance() 
public static MessagePropertiesBuilder fromProperties(MessageProperties properties) 
public static MessagePropertiesBuilder fromClonedProperties(MessageProperties properties) 
新消息属性将使用默认值进行初始化
builder会使用提供的properties对象进行初始化，build() 方法也会返回参数properties对象．
参数的属性会拷贝到新的MessageProperties对象中.
在AmqpTemplate的RabbitTemplate 实现中, 每个send() 方法的重载版本都接受一个额外的CorrelationData对象.
当启用了发布者确认时，此对象会在3.1.4, “AmqpTemplate”的回调中返回.这允许发送者使用确认(ack或nack)来关联发送的消息．
发布者返回
当模板的mandatory 属性为true时，返回消息将由 Section 3.1.4, “AmqpTemplate”描述的回调来返回.
从１.４版本开始，RabbitTemplate 支持 SpEL mandatoryExpression 属性,它将对每个请求消息进行评估，作为根评估对象来解析成布尔值. Bean引用,如"@myBean.isMandatory(#root)" 可用在此表达式中．
发布者返回内部也可用于RabbitTemplate 的发送和接收操作中. 参考the section called “Reply Timeout” 来了解更多信息.
批量
从１.４.２版本开始，引入了BatchingRabbitTemplate.它是RabbitTemplate 的子类，覆盖了send 方法，此方法可根据BatchingStrategy来批量发送消息; 只有当一个批次完成时才会向RabbitMQ发送消息。
public interface BatchingStrategy {

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

	Date nextRelease();

	Collection<MessageBatch> releaseBatches();

}

• batchSize - 发送前一个批次中消息的数量
• bufferLimit - 批量消息的最大大小;如果超过了此值，它会取代batchSize, 并导致要发送的部分批处理
• timeout - 当没有新的活动添加到消息批处理时之后，将发送部分批处理的时间(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 接收总是比发送稍显复杂.有两种方式来接收Message. 最简单的选择是在轮询方法调用中一次只接收一个消息. 更复杂的更常见的方法是注册一个侦听器，按需异步的接收消息。
在下面两个子章节中，我们将看到这两种方法的示例.

AmqpTemplate 自身可用来轮询消息接收.默认情况下，如果没有可用消息,将会立即返回 null;它是无阻塞的.
从1.5版本开始，你可以设置receiveTimeout,以毫秒为单位, receive方法会阻塞设定的时间来等待消息.小于０的值则意味着无限期阻塞 (或者至少要等到与broker的连接丢失).
1.6版本引入了receive 方法的变种，以允许在每个调用上都可设置超时时间.

警告

由于接收操作会为每个消息创建一个新的QueueingConsumer,这种技术并不适用于大容量环境，可考虑使用异步消费者，或将receiveTimeout 设为０来应对这种情况．

这里有四个简单可用的receive 方法.同发送方的交换器一样, 有一种方法需要直接在模板本身上设置的默认队列属性, 还有一种方法需要在运行接受队列参数.
版本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;

与发送消息的情况类似， AmqpTemplate 有一些便利的方法来接收POJOs 而非Message 实例, 其实现可提供一种方法来定制MessageConverter 以用于创建返回的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 方法同步接收，处理，以及回应消息:

<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 实现会负责receive 和 reply 阶段.在大多数情况下，如果有必要，你只需要提供ReceiveAndReplyCallback 的实现来为收到的消息执行某些业务逻辑或为收到的消息构建回应对象.
注意，ReceiveAndReplyCallback 可能返回null. 在这种情况下，将不会发送回应，receiveAndReply 的工作类似于receive 方法. 这允许相同的队列用于消息的混合物，其中一些可能不需要答复。

自动消息(请求和应答)转换只能适应于提供的回调不是ReceiveAndReplyMessageCallback 实例的情况下- 它提供了一个原始的消息交换合同。

ReplyToAddressCallback 只在这种情况中有用，需要根据收到的信息通过自定义逻辑来决定replyTo 地址，并在ReceiveAndReplyCallback中进行回应的情况. 默认情况下，请求消息中的 replyTo 信息用来路由回复.

下面是一个基于POJO的接收和回复…​

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!");
}

异步消费者

对于异步消息接收, 会涉及到一个专用组件(不是AmqpTemplate).此组件可作为消息消费回调的容器.
稍后，我们会讲解这个容器和它的属性,但首先让我们来看一下回调，因为这里是你的应用程序代码与消息系统集成的地方. MessageListener 接口:

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

如果出于任何理由，你的回调逻辑需要依赖于AMQP Channel实例,那么你可以使用ChannelAwareMessageListener. 它看起来是很相似的，但多了一个额外的参数：

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

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

你也可以继承适配器，并实现getListenerMethodName()方法(基于消息来动态选择不同的方法). 这个方法有两个参数：originalMessage 和extractedMessage, 后者是转换后的结果.默认情况下，需要配置SimpleMessageConverter ;
参考the section called “SimpleMessageConverter” 来了解更多信息以及其它转换器的信息.

从1.4.2开始，原始消息包含consumerQueue 和 consumerTag 属性，这些属性可用来确定消息是从那个队列中收到的．

从1.5版本开始，你可以配置消费者queue/tag到方法名称的映射(map)以动态选择要调用的方法.如果map中无条目，我们将退回到默认监听器方法．

你已经看过了消息监听回调上的各种各样的选项，现在我们将注意力转向容器. 基本上，容器处理主动(active)的职责,这样监听器回调可以保持被动(passive）. 容器是“生命周期”组件的一个例子。
它提供了启动和停止的方法.当配置容器时，你本质上缩短了AMQP Queue和 MessageListener 实例之间的距离.你必须提供一个ConnectionFactory 的引用，队列名称或队列实例．
下面是使用默认实现SimpleMessageListenerContainer 的最基础的例子:

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

作为一个主动组件, 最常见的是使用bean定义来创建监听器容器，这样它就可以简单地运行于后台.这可以通过XML来完成:

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

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

</rabbit:listener-container>

或者你可以@Configuration 风格:

@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支持消费者优先级了(参考 Using Consumer Priorities with RabbitMQ). 

这可以通过在消费者设置x-priority 参数来启用． 

SimpleMessageListenerContainer 现在支持设置消费者参数:

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>

从１.３版本开始，容器监听的队列可在运行时进行修改，参考 Section 3.1.18, “Listener Container Queues”.

当容器配置为监听auto-delete 队列或队列有x-expires 选项或者broker配置了Time-To-Live 策略,队列将在容器停止时(最后的消费者退出时)由broker进行删除.
在１.３版本之前，容器会因队列缺失而不能重启; 当连接关闭／打开时，RabbitAdmin 只能自动重新声明队列.

从1.3版本开始, 在启动时，容器会使用RabbitAdmin 来重新声明缺失的队列．

您也可以使用条件声明(the section called “Conditional Declaration”) 与auto-startup="false" 来管理队列的延迟声明，直到容器启动.

<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" />

在这种情况下，队列和交换器是由 containerAdmin 来声明的，auto-startup="false" 因此在上下文初始化期间不会声明元素.同样，出于同样原因，容器也不会启动.当容器随后启动时，它会使用containerAdmin引用来声明元素.

批量消息

消费者失败事件

• container - 消费者经历问题的监听容器．
• reason - 失败的文本原因。
• fatal - 一个表示失败是否是致命的boolean值;对于非致命异常,容器会根据retryInterval值尝试重新启动消费者.
• throwable -捕捉到的Throwable.