Spring AMQP参考手册

71 阅读 0 评论 47 点赞

我是靠谱客的博主悦耳猫咪，最近开发中收集的这篇文章主要介绍Spring AMQP参考手册，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

一、快速入门

Maven依赖声明

<dependency>
    <groupId>org.springframework.amqp</groupId>
    <artifactId>spring-rabbit</artifactId>
    <version>1.3.5.RELEASE</version>
</dependency>

Java Config方式的示例程序

ApplicationContext context = new AnnotationConfigApplicationContext(RabbitConfiguration.class);
AmqpTemplate template = context.getBean(AmqpTemplate.class);
template.convertAndSend("myqueue", "foo");
String foo = (String) template.receiveAndConvert("myqueue");

配置类：

@Configuration
public class RabbitConfiguration {
@Bean
public ConnectionFactory connectionFactory() {
CachingConnectionFactory connectionFactory =
new CachingConnectionFactory("localhost");
return connectionFactory;
}
@Bean
public AmqpAdmin amqpAdmin() {
return new RabbitAdmin(connectionFactory());
}
@Bean
public RabbitTemplate rabbitTemplate() {
return new RabbitTemplate(connectionFactory());
}
@Bean
public Queue myQueue() {
return new Queue("myqueue");
}
}

三、使用AMQP

3.1 AMQP概念

Spring AMQP包含几个模块，每个模块都对应一个Jar包，它们是：spring-amqp, spring-rabbit和spring-erlang。

spring-amqp：包含了核心的“AMQP模型”，该模型对AMQP对象进行了概括，使得其不依赖于特定的AMQP broker实现

方案或特定的客户端程序库。然后，这些抽象的模型由特定的broker来实现，例如“spring-rabbit”。

注：关于ErLang

Erlang是一种通用的面向并发的编程语言，它由瑞典电信设备制造商爱立信所辖的CS-Lab开发，目的是创造一种可以

应对大规模并发活动的编程语言和运行环境。Erlang是运行于虚拟机的解释性语言，但是现在也包含有本地代码编译

器。Erlang属于多重范型编程语言，涵盖函数式、并发式及分布式。

消息（Message）

AMQP本来是传输字节流的信息，因此在以前Spring并没有定义Message类，但是最近Spring AMQP定义了一个Message类

，这样就能够在一个实例中封装消息体（body）和属性（properties），从而使API变得简单。Message的定义非常简

单，就是包含了一个字节数组：body和MessageProperties。

Exchange

Exchange代表消息提供者想要发送消息的目的地。每个Exchange包含在Broker的虚拟主机中，有一个唯一的名字和其

他属性。

Routing Key：路由关键字

Exchange有其所能支持的交换类型：

Direct：Direct Exchange允许Queue以固定的routing key（通常是Queue的名字）绑定
Topic：Topic Exchange允许使用通配符进行绑定
Fanout：不使用任何routing key，任何Queue都可以绑定
Headers：未知
System：未知

注意：AMQP规范要求每个Broker都有一个默认的，没有名称的Direct Exchange。所有的Queue都将使用他们的Queue

Name绑定到这个Exchange。

Queue

Queue代表消息使用者接收消息的组件。

Binding

消息提供者将消息发送到Exchange，而消息接收者是从Queue接收消息，Binding定义了两者（Exchange和Queue）之间

的关系。

绑定的示例：

1、使用固定的routing key将Queue绑定到Direct Exchange

new Binding(someQueue, someDirectExchange, "foo.bar")

2、使用通配符绑定到一个TopicExchange

new Binding(someQueue, someTopicExchange, "foo.*")

3、绑定到Fanous Exchange

new Binding(someQueue, someFanoutExchange)

4、使用流API的样式

Binding b = BindingBuilder.bind(someQueue).to(someTopicExchange).with("foo.*");

3.2 连接和资源管理

虽然AMQP模型是抽象的通用的概念，但是当我们实现资源管理时，其细节是和具体的broker相关的。在本文中，假定

以使用RabbitMQ作为实现方案。

RabbitMQ管理连接的核心组件是ConnectionFactory接口。ConnectionFactory的作用是提供Connection的实例（在

RabbitMQ方案中，Connection就是com.rabbitmq.client.Connection的封装）。

ConnectionFactory的唯一实现类是CachingConnectionFactory，该类默认情况下将建立一个在应用程序间共享的

Connection代理（connection proxy）。

AMQP的消息机制进行工作的处理单元（unit of work）是Channel（类似于JMS中的Connection和Session的概念），这

种机制使得在应用程序间共享连接是可能的。

Connection提供了createChannel方法，而CachingConnectionFactory会对所创建的Channel进行缓存，这些缓存将考

虑到Channel的事务属性。

Spring AMQP 1.3版本后，不仅Channel在CachingConnectionFactory是缓存的，Connection也是缓存起来的。在一些

情况下需要使用单独的连接：例如HA或集群的情况下。

【重要】
1、如果CachingConnectionFactory的cache mode为Connection，则不支持自动创建Queue
2、目前rabbitmq-client库为每个Connection创建5个线程，如果Connection太多，则这种处理方式可能不太适合。这

时你需要为CachingConnectionFactory设置一个自定义的Executor。如果每个连接建立多个Channel，则线程池的大小

会影响并发性能，因此最好使用可变的线程池

配置ConnectionFactory

最简单的方法是使用rabbitmq的namespace：

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

如果你还想定制更多的内容，例如将Channel Cache Size从默认的1变为25：

或者，直接在namespace中修改：

<rabbit:connection-factory id="connectionFactory" channel-cache-size="25"/>

默认情况下，cache mode为CHANNEL，你也可以将Cache Mode改为Connection：

<rabbit:connection-factory id="connectionFactory" cache-mode="CONNECTION" connection-cache-size="25"/>

指定host name和端口

<rabbit:connection-factory id="connectionFactory" host="somehost" port="5672"/>

集群情况下

<rabbit:connection-factory id="connectionFactory" addresses="host1:5672,host2:5672"/>

如果你还要指定其他的特性，可以自己重定义一个ConnectionFactory：

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

在上例中，我们引入一个SimpleRoutingConnectionFactory，它可以在运行时通过lookupKey，从多个

ConnectionFactory中确定其中一个。

消息确认和消息回复（Publisher Confirms and Returns）

当CachingConnectionFactory的publisherConfirms和publisherReturns属性设为真时，该Factory支持消息确认和消息回复。在这种情况下，该Factory创建的Channel会由PublisherCallbackChannel进行封装，该类提供一些机制允许消息提供者处理消息确认或消息回复到达时的回调。客户端应用程序可以注册PublisherCallbackChannel.Listener来处理回调。

关于此方面的更多信息，请参见RabbitMQ开发团队的BLOG http://www.rabbitmq.com/blog/2011/02/10/introducing-publisher-confirms/

注：该文章对于如何更快地发送消息进行了一个技术分析，其举例说，在一些业务场景下需要确保发送的消息“基本上”不丢失。如果按照传统的做法，发送操作包含在事务处理中，每次发送都需要提交事务，则发送10000条消息大约需要4分钟。实际上，业务逻辑只需要“知道哪些没有发送成功”，因此完全可以利用Confirms机制，异步地来知道哪些没有发送成功，此项改进使得发送时间缩短到2秒，如下所示：

ch.setConfirmListener(new ConfirmListener() {
    public void handleAck(long seqNo, boolean multiple) {
        if (multiple) {
            unconfirmedSet.headSet(seqNo+1).clear();
        } else {
            unconfirmedSet.remove(seqNo);
        }
    }
    public void handleNack(long seqNo, boolean multiple) {
        // handle the lost messages somehow
    }
});

3.3 AmqpTemplate

和其他SpringFramework机制一样，AmqpTemplate提供了发送和接收消息的方法，从某方面来说，它和其他xxxxTemplate并无不同，从另一方面来说，它的具体实现却和AMQP的实现方案紧密相关。和JMS不一样的是，JMS本身就规定了各种接口，而AMQP只是规定了线路层的协议。AMQP的不同实现方案都可以有自己的具体接口，当前，只有一个实现方案：RabbitTemplate。在本文中，我们可以经常看到使用AmqpTemplate，但是当你研究其底层实现时，你会发现其实例是RabbitTemplate。

使用重发功能（Add Retry Capabilities）

Spring AMQP 1.3版本后，我们可以配置RabbitTemplate使用RetryTemplate来实现重发功能。默认情况下，使用SimpleRetryPolicy在抛出异常前会重试三次，在下例中，使用了一个间隔时间指数性增长的策略来规定如何重试：

使用XML：

<rabbit:template id="template" connection-factory="connectionFactory" retrytemplate="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;
}

注：Retry功能是Spring提供的一个Project，目的是提供可声明的重试机制，参见：https://github.com/spring-projects/spring-retry

消息确认和消息回复（Publisher Confirms and Returns）

RabbitTemplate的实现方案支持消息确认和消息回复，对于消息回复，AmqpTemplate的mandatory属性必须设为真，同时CachingConnectionFactory的publisherReturns属性也必须设为真。

客户端应用程序将调用template的setReturnCallback(ReturnCallback
callback)注册回调函数。该回调函数必须实现这个方法：

void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey);

每个RabbitTemplate只允许一个ReturnCallback。

对于消息确认（也称为Publisher Acknowledgements），CachingConnectionFactory的publisherConfirms属性需要设为真，同样，客户端应用程序通过注册ConfirmCallback来处理回调，该回调必须实现方法：

void confirm(CorrelationData correlationData, boolean ack);

CorrelationData是消息提供者发送消息时的一个关联数据，比如Original Message Id等。同样，每个RabbitTemplate也只允许一个ConfirmCallback。

3.4 发送消息（Sending messages）

发送消息时，我们可以使用下列三个方法之一：

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（路由关键字）。如果这两个参数在template中已经指定，则可以使用上面两个简化的方法，例如：

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

我们在考虑exchange和routing key的时候，不管在方法中还是在template中，我们总是会想到要去设这些参数，而事实上，即使你哪也不设，在发送过程中总会使用合适的默认值（空字符串）。对于routing key来说，Queue绑定到Exchange时可以使用空字符串，而且对于Fanous Exchange也不需要routing key。对于Exchange，AMQP规范就规定了总会有一个名字为空的Default Exchange。

既然所有的Queue都会以Queue Name作为绑定值绑定到这个默认的Direct Exchange，那么对于点对点通信，我们只需要使用第二种方法，使用Queue Name作为routing key即可：

RabbitTemplate template = new RabbitTemplate(); // using default no-name Exchange
template.send("queue.helloWorld", new Message("Hello World".getBytes(), someProperties));

消息生成器API

自1.3版本后，Spring AMQP提供了MessageBuilder和MessagePropertiesBuilder两个生成器。它们为创建消息提供了流式的方法，例如：

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中的所有属性都可以通过这两个生成器进行设置，另外，还可以通过set*IfAbsent()和set*IfAbsentOrDefault()，在属性不包含值或者只包含默认值的情况下进行设置。

另外，创建消息时也不用白手起家，我们可以通过下面的五个静态方法使MessageBuilder具备初始内容：

public static MessageBuilder withBody(byte[] body)
直接使用该数组作为消息的body

public static MessageBuilder withClonedBody(byte[] body)
直接使用该数组的复制作为消息的body

public static MessageBuilder withBody(byte[] body, int from, int to)
复制该数组的一部分作为消息的body

public static MessageBuilder fromMessage(Message message)
新的消息和参数所提供的消息共用body，但是MessageProperties是复制的

public static MessageBuilder fromClonedMessage(Message message)
复制参数所提供的消息的body和MessageProperties作为新消息

我们还可以下列方法使MessagePropertiesBuilder具备初始内容：

public static MessagePropertiesBuilder newInstance()
使用默认的properties初始化消息

public static MessagePropertiesBuilder fromProperties(MessageProperties properties)
该生成器的build()方法将直接返回参数所指定的MessageProperties对象

public static MessagePropertiesBuilder fromClonedProperties(MessageProperties properties)
复制参数所指定的MessageProperties对象的内容

消息确认

对于RabbitTemplate，每个send()方法都有一个重载的方法，输入CorrelationData。

消息回复

参见3.3 AmqpTemplate

3.5 接收消息

消息接收通常比发送复杂一点，其中之一的原因是有同步和异步两种方式来接收消息。简单一点的方式是同步的轮询，另一种稍复杂的方式是注册一个listener，当消息到达时通过listener异步地接收。接下来我们将探讨这两种方式。

轮询的消息使用者（Polling Consumer）

AmqpTemplate可以直接用于轮询方式的消息接收。如果没有消息抵达，接收时将立刻返回null值，不会阻塞线程。例如：

Message receive() throws AmqpException;
Message receive(String queueName) throws AmqpException;

AmqpTemplate也提供了另一种方式接收POJO对象（而不是Message），并可以设置MessageConverter将消息转换到POJO对象：

Object receiveAndConvert() throws AmqpException;
Object receiveAndConvert(String queueName) throws AmqpException;

从1.3版本开始，如同sendAndReceive方法一样，接收者也可以使用receiveAndReply来同步地接收、处理并回复消息。

在大多数情况下，你只需要使用ReceiveAndReplyCallback来提供有关接收消息和回复消息（或POJO）的业务逻辑。需要注意的是，ReceiveAndReplyCallback可能返回null，在这种情况下，就像receive方法一样。这允许我们使用同一个Queue来混合处理那些需要回复和不需要回复的消息。

默认情况下，回复消息的目的地址来自于请求消息，如果需要指定，则可以通过ReplyToAddressCallback来进行。

下面是基于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!");
}

异步的消息使用者（Asynchronous Consumer）

对于异步的消息接收，需要使用另一个组件（不是AmqpTemplate），该组件是一个Listener容器。我们随后将介绍这个容器，但是在此之前，我们先来看看回调过程。实现回调的最简单方式是实现MessageListener接口：

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

如果你的Listener需要知道当前的Channel，则可以使用ChannelAwareMessageListener：

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

如果你想要将业务逻辑和消息处理严格分开，则可以引入Adapter：

MessageListener listener = new MessageListenerAdapter(somePojo);

接下来我们介绍Listener容器，该容器的首要职责是“激活”不同的listener，这样我们的listener就能保持被动接收的模式。该容器是一个具有生命周期（lifecycle）的组件，它提供了启动和停止的方法。当配置该容器时，我们需要建立AMQP的Queue和Listener之间的联系：需要配置Listener使其知道需要从ConnectionFactory的哪个Queue中接收消息。

下面是一个使用默认的SimpleMessageListenerContainer的简单方案：

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

作为一个组件，listener容器通常也需要作为bean定义到应用程序上下文中，以便它能够在后台启动并监听消息：

<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() {
  return new MessageListener() {
   public void onMessage(Message message) {
    System.out.println("received: " + message);
   }
  };
}
}

自RabbitMQ 3.2开始，该broker允许指定消息使用者的优先级（参见http://www.rabbitmq.com/blog/2013/12/16/using-consumer-priorities-with-rabbitmq/），如下所示：

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

或

<rabbit:listener-container connection-factory="rabbitConnectionFactory">
<rabbit:listener queues="some.queue" ref="somePojo" method="handle" priority="10" />
</rabbit:listener-container>

从1.3版本开始，Spring AMQP允许在运行时改变容器所监听的Queue，参见3.14 Listener Container Queues。

临时队列（auto-delete queue）

当Listener容器监听一个临时队列时，则在容器停止后，该队列将被删除，在1.3版本前，该容器会由于原本所监听的队列已经不存在而无法重新启动。1.3版本后，Listener容器启动时会通过RabbitAdmin重定义任何不存在的队列，因此不再存在这个问题。

你也可以使用条件声明来推迟队列的建立，直到容器启动时再建立队列。

3.6 Message Converters

AmqpTemplate定义了一些方法用于消息的转换委托给一个MessageConverter。MessageConverter的定义相当简单：

public interface MessageConverter {
Message toMessage(Object object, MessageProperties messageProperties) throws MessageConversionException;
Object fromMessage(Message message) throws MessageConversionException;
}

请注意对于toMessage来说，由于POJO通常对应的是消息的body，因此还会提供一个MessageProperties给转换器。

添加了MessageConverter之后，我们可以调用相应的convertAndSend()和receiveAndConvert()方法来发送和接收消息，这些方法直接使用POJO，而转换的事情则交给了MessageConverter。
在异步的情况下，MessageListenerAdapter也支持使用MessageConverter。

SimpleMessageConverter

MessageConverter的默认实现方案是SimpleMessageConverter。如果你不指定MessageConverter，RabbitTemplate就会使用该Converter，该Converter支持文本、可序列化的Java对象和字节数组。

如果content type是文本（"text/plain"），它会检查字符编码，如果消息中没有指定，则会默认以UTF-8进行识别，如果你不希望这样，可以自己指定defaultChartset属性。

如果content type是"application/x-java-serialized-object"，SimpleMessageConverter将试图通过反序列化将字节数组转换为Java对象。

但是，我们不要认为这个过程一定是使用Java序列化机制——这样不利于发送者和接收者之间解耦。AMQP作为线路层（wire-level）的协议，必然也会给我们开发带来一些不利——我们无法约束其一定使用Java序列化机制。在接下来的章节，我们将探讨如果不使用Java序列化机制来传输那些复杂对象（rich domain object content）。

对于其他类型，SimpleMessageConverter直接将消息body作为字节数组返回。

转换至Message

SimpleMessageConverter可以将字节数组、字符串和可序列化对象转换为Message并设置合适的类型，对于其他内容，消息的body则为null。

JsonMessageConverter和Jackson2JsonMessageConverter

如上文所述，我们不建议依赖于Java序列化机制，因此我们更推荐使用JSON作为序列化机制，它具备更好的灵活性和跨语言和平台的一致性。目前有两个替代默认SimpleMessageConverter的机制，可以配置到任何RabbitTemplate实例中。

JsonMessageConverter：使用org.codehaus.jackson 1.x库
Jackson2JsonMessageConverter：使用com.fasterxml.jackson 2.x库

其配置方法如下所示：

在上例中，Json对象的类型信息将保存至MessageProperties中（或自MessageProperties中获取）。如果消息中不包含

类型信息，而你知道其类型，可以指定defaultType属性：

MarshallingMessageConverter

消息转换的另一个方案是MarshallingMessageConverter，它由Spring OXM库实现，配置方法类似：

3.7 请求/回复模式的消息处理

AmqpTemplate提供了一系列sendAndReceive方法，其所用到的参数和单程的发送方法一样，这些方法在请求/回复这样的场景中比较合适。这些方法可以配置“reply-to”属性，并且可以通过注册到一个单独的队列的Listener来接收回复消息。这些情况下同样可以使用MessageConverter，对应的方法为convertSendAndReceive。

默认情况下，每个reply都会保存到一个临时创建的队列，然而在AmqpTemplate中配置一个单独的队列会更有效率，并且可以设置一些参数。在这种情况下，你仍然需要提供一个<reply-listener>标签，该标签会为该reply queue创建listener container，你可以为该container设置除了connection-factory和message-converter以外的所有属性（这两个属性是从AmqpTemplate中继承的）：

<rabbit:template id="amqpTemplate" connection-factory="connectionFactory" reply-queue="replies">
<rabbit:reply-listener/>
</rabbit:template>

虽然reply queue的container和template共享Connection Factory，但是他们不会共享使用同一个Channel，因此消息回复不会在同一个事务中。

Reply Queue的消息对应（Message Correlation With A Reply Queue）

当使用固定的reply queue时，我们需要提供关联数据，这样消息回复可以对应到消息请求。默认情况下，correlationId属性保存了这个关联数据，你也可以使用correlation-key属性来指定其他字段作为关联数据。

注：Spring AMQP 1.1版本使用了一个spring_reply_correlation字段作为关联数据，如果你的应用程序需要与1.1版本

的应用程序进行交互，则需要设置spring_reply_correlation字段。

如果你通过Spring的POM将RabbitTempalte定义为<bean>（而不是<rabbit:template>标签），或者使用配置类定义，或者动态创建template，则你必须自己定义并设置reply listener container，否则该template将不会接收到消息。

重要：当你自己设置reply listener和template时，必须要保证template的reply queue和container的queue是一样的。下面的例子说明了如何自己设置这些bean：

或

@Bean
public RabbitTemplate amqpTemplate() {
RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory());
rabbitTemplate.setMessageConverter(msgConv());
rabbitTemplate.setReplyQueue(replyQueue());
rabbitTemplate.setReplyTimeout(60000);
return rabbitTemplate;
}
@Bean
public SimpleMessageListenerContainer replyListenerContainer() {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setConnectionFactory(connectionFactory());
container.setQueues(replyQueue());
container.setMessageListener(amqpTemplate());
return container;
}
@Bean
public Queue replyQueue() {
return new Queue("my.reply.queue");
}

Spring Remoting with AMQP

Spring Framework支持基本的远程化处理，即允许以多种传输方式进行RPC调用。Spring AMQP通过在客户端提供AmqpProxyFactoryBean和在服务器端提供AmqpInvokerServiceExporter来提供类似机制，底层仍然使用RabbitTemplate和MessageListener来进行传输。详情略。

3.8 配置Broker

AMQP规范规定了如何在broker上配置Queue、Exchange、Binding。这些操作在Spring AMQP中通过AmqpAdmin接口进行定义，具体的实现方案是RabbitAdmin。

AmqpAdmin接口定义如下：

public interface AmqpAdmin {
// Exchange Operations
void declareExchange(Exchange exchange);
void deleteExchange(String exchangeName);
// Queue Operations
Queue declareQueue();
String declareQueue(Queue queue);
void deleteQueue(String queueName);
void deleteQueue(String queueName, boolean unused, boolean empty);
void purgeQueue(String queueName, boolean noWait);
// Binding Operations
void declareBinding(Binding binding);
void removeBinding(Binding binding);
Properties getQueueProperties(String queueName);
}

declareQueue()在broker中定义一个新的queue，名字自动生成，其他属性为exclusive=true, autoDelete=true, durable=false。
declareQueue(Queue queue)将在broker中添加指定的Queue，返回该Queue的名字。如果Queue没有名字，则会为该Queue生成一个名字并返回给调用者。但是在application context中以xml方式定义queue，则不适用于这种情况，如果通过“<rabbit:queue/>”定义queue，又没有指定名字，则broker总是创建匿名队列（名称是一个自动生成的UUID，durable=false, exlusive=true, autoDelete=true）。因为当连接失效时，匿名队列的名字总是会变化的。

声明式的队列必须有固定的名字，以便在其他地方引用，例如：

<rabbit:listener-container>
<rabbit:listener ref="listener" queue-names="#{someQueue.name}" />
</rabbit:listener-container>

参见章节“Automatic Declaration of Exchanges, Queues and Bindings”。

RabbitMQ的实现方案是RabbitAdmin，配置示例如下：

<rabbit:connection-factory id="connectionFactory"/>
<rabbit:admin id="amqpAdmin" connection-factory="connectionFactory"/>

当CachingConnectionFactory的cache mode为CHANNEL时（默认），RabbitAdmin将在connection启动时在application context中自动定义Queue、Exchange和Binding。下面是配置示例：

<rabbit:queue id="tradeQueue"/>
<rabbit:queue id="marketDataQueue"/>
<fanout-exchange name="broadcast.responses" xmlns="http://www.springframework.org/schema/rabbit">
<bindings>
<binding queue="tradeQueue"/>
</bindings>
</fanout-exchange>
<topic-exchange name="app.stock.marketdata" xmlns="http://www.springframework.org/schema/rabbit">
<bindings>
<binding queue="marketDataQueue" pattern="${stocks.quote.pattern}"/>
</bindings>
</topic-exchange>

在上例中，我们使用了匿名队列（请注意这里提供的是id，而不是name），匿名队列的名字由spring framework生成，而不是broker。其他地方通过id作为标识来引用这个队列。当然我们也可以使用name，name同样可以作为标识，如下所示：

<rabbit:queue name="stocks.trade.queue"/>

提示：你可以同时提供id和name，这样你可以通过id来引用这个queue，而名字可以使用另外的值。这样你可以在queue name中使用占位符或SpEL表达式。当然如果通过<bean>来定义queue，则name必须为确定的值。

也可以为Queue配置更多的属性，属性的值默认情况下被当成字符串对待，如果是其他的类型，则需要提供类型：

<rabbit:queue name="withArguments">
<rabbit:queue-arguments>
<entry key="x-message-ttl" value="100" value-type="java.lang.Long"/>
<entry key="x-ha-policy" value="all"/>
</rabbit:queue-arguments>
</rabbit:queue>

注：以上定义方式需要Spring Framework 3.2或以上。

重要

RabbitMQ不允许定义相冲突的属性。例如，如果已经有一个queue存在，其no-time-to-live属性（即不允许设置ttl），然而，你试图指定key="x-message-ttl" value="100"，则会产生异常。

默认情况下，RabbitAdmin在遇到任何异常时，都会停止声明任何资源，这样可能导致一些关联错误（downstream issues）——比如listener container初始化失败，其原因是其所依赖的一个queue没有被声明（其声明顺序位于那个出错的queue之后）。

当然你也可以将RabbitAdmin的ignore-declaration-failures设为真。这样RabbitAdmin会继续声明余下的资源，并记录该异常。

Spring AMQP 1.3后，可以配置HeadersExchange，通过Header的匹配关系将Queue绑定到Exchange，例如：

<rabbit:headers-exchange name="headers-test">
<rabbit:bindings>
  <rabbit:binding queue="bucket">
   <rabbit:binding-arguments>
    <entry key="foo" value="bar"/>
    <entry key="baz" value="qux"/>
    <entry key="x-match" value="all"/>
   </rabbit:binding-arguments>
  </rabbit:binding>
</rabbit:bindings>
</rabbit:headers-exchange>

如果想要了解如何通过Java Config方式来配置AMQP，可以参见stock示例应用程序，其使用了AbstractStockRabbitConfiguration来配置基本的信息，然后通过RabbitClientConfiguration和RabbitServerConfiguration配置客户端和服务器端的额外信息，如下所示：

@Configuration
public abstract class AbstractStockAppRabbitConfiguration {
@Bean
public ConnectionFactory connectionFactory() {
  CachingConnectionFactory connectionFactory = new CachingConnectionFactory(
    "localhost");
  connectionFactory.setUsername("guest");
  connectionFactory.setPassword("guest");
  return connectionFactory;
}

@Bean
public RabbitTemplate rabbitTemplate() {
  RabbitTemplate template = new RabbitTemplate(connectionFactory());
  template.setMessageConverter(jsonMessageConverter());
  configureRabbitTemplate(template);
  return template;
}

@Bean
public MessageConverter jsonMessageConverter() {
return new JsonMessageConverter();
}

@Bean
public TopicExchange marketDataExchange() {
return new TopicExchange("app.stock.marketdata");
}
// additional code omitted for brevity
}

@Configuration
public class RabbitServerConfiguration extends
AbstractStockAppRabbitConfiguration {
@Bean
public Queue stockRequestQueue() {
return new Queue("app.stock.request");
}
}

Server端配置类和基础配置类形成一个这样的配置结果：在broker启动时，TopicExchange和一个queue被创建，但是并没有queue绑定到TopicExchange（这个将在客户端应用程序中实现），而stock request queue将自动绑定到default exchange。

我们来看看客户端的配置类，这更有意思：

@Configuration
public class RabbitClientConfiguration extends
AbstractStockAppRabbitConfiguration {
@Value("${stocks.quote.pattern}")
private String marketDataRoutingKey;

@Bean
public Queue marketDataQueue() {
return amqpAdmin().declareQueue();
}

/**
* Binds to the market data exchange. Interested in any stock quotes that
* match its routing key.
*/
@Bean
public Binding marketDataBinding() {
return BindingBuilder.bind(marketDataQueue()).to(marketDataExchange())
.with(marketDataRoutingKey);
}
// additional code omitted for brevity
}

客户端应用程序定义了另一个queue，将其绑定到market data exchange，其routing通配字符串保存在外部配置文件中。

条件式声明（Conditional Declaration）

默认情况下，所有资源（queue、exchange、bindings）由RabbitAdmin在application context中自动定义。从Spring AMQP 1.2开始，我们可以有条件地定义这些资源，这在我们需要将应用程序连接到多个broker，并且指定哪个资源应该在哪个broker中定义时，变得非常有用。

这些资源都实现了Declarable接口，该接口有两个方法：shouldDeclare()和getDeclaringAdmins()，RabbitAdmin通过这两个方法来决定是否需要基于它的连接来定义相关的资源。这些信息可以通过namespace提供，例如：

<rabbit:admin id="admin1" connection-factory="CF1" />
<rabbit:admin id="admin2" connection-factory="CF2" />
<rabbit:queue id="declaredByBothAdminsImplicitly" />
<rabbit:queue id="declaredByBothAdmins" declared-by="admin1, admin2" />
<rabbit:queue id="declaredByAdmin1Only" declared-by="admin1" />
<rabbit:queue id="notDeclaredByAny" auto-declare="false" />
<rabbit:direct-exchange name="direct" declared-by="admin1, admin2">
<rabbit:bindings>
<rabbit:binding key="foo" queue="bar"/>
</rabbit:bindings>
</rabbit:direct-exchange>

当auto-declare属性为真（默认），并且没有指定declared-by属性时，所有的RabbitAdmin将定义该资源。当然，如果RabbitAdmin的auto-startup属性为假，则该RabbitAdmin就不会这么做（默认auto-startup为真）。

同样，我们也可以使用Java Config方式来声明，在下例中，所有资源由admin来定义，admin2不定义这些内容：

@Bean
public RabbitAdmin admin() {
  RabbitAdmin rabbitAdmin = new RabbitAdmin(cf1());
  rabbitAdmin.afterPropertiesSet();
  return rabbitAdmin;
}

@Bean
public RabbitAdmin admin2() {
  RabbitAdmin rabbitAdmin = new RabbitAdmin(cf2());
  rabbitAdmin.afterPropertiesSet();
  return rabbitAdmin;
}

@Bean
public Queue queue() {
  Queue queue = new Queue("foo");
  queue.setAdminsThatShouldDeclare(admin());
  return queue;
}

@Bean
public Exchange exchange() {
  DirectExchange exchange = new DirectExchange("bar");
  exchange.setAdminsThatShouldDeclare(admin());
  return exchange;
}

@Bean
public Binding binding() {
  Binding binding = new Binding("foo", DestinationType.QUEUE, exchange()
    .getName(), "foo", null);
  binding.setAdminsThatShouldDeclare(admin());
  return binding;
}

3.9 异常处理（Exception Handling）

使用RabbitMQ的Java程序的许多操作都会抛出可检查的异常（checked exception）。例如许多情况下程序可能遇到IOExceptions异常。RabbitTemplate、SimpleMessageListenerContainer和其他Spring AMQP组件将捕获到这些异常并按照AMQP的异常体系将其转换成合适的异常。这些异常定义在org.springframework.amqp包中，其异常体系的基础类是AmqpException。

当一个Listener检测到异常，它会将其封装为一个ListenerExecutionFailedException异常，同时消息将被拒绝并且由broker重新加入队列。将defaultRequeueRejected属性设为false将导致消息被丢弃（或者转储到一个专门的Exchange（dead letter exchange）），此时抛出的异常为AmqpRejectAndDontRequeueException。

然而对于有一种情形，Listener无法妥善处理。当消息无法被转换时，由于程序还没有运行到你的代码，因此在defaultRequeueRejected为真时（默认情况），消息就会被不断的重发。

在1.3.2版本前，用户需要自己为该Listener写一个ErrorHandler，来防止出现消息格式错误时不断重发的情形，自1.3.2版本后，Listener的ErrorHandler为ConditionalRejectingErrorHandler，它会拒绝那些格式错误的消息并抛出MessageConversionException异常（消息不会再被加入队列）。新的ErrorHandler中可以配置FatalExceptionStrategy，这样用户在消息被拒绝时，可以使用自己的处理规则，例如由Spring Retry委托至BinaryExceptionClassifier进行处理。另外，ListenerExecutionFailedException现在也有一个failedMessage属性帮助用户应用程序来决定如何处理那些出错的消息。如果FatalExceptionStrategy.isFatal()方法返回真，那么ErrorHandler会抛出AmqpRejectAndDontRequeueException异常，默认情况下，FatalExceptionStrategy的策略是记录一条警告消息。

3.10 事务处理（Transactions）

Spring Rabbit框架支持在同步和异步场景下的可声明式的自动事务管理，它使用的一系列语义和其他Spring事务程序类似，其他很多通用消息处理方案是难以实现这样的功能的。

有两种方式可以向Spring Framework标记事务。首先，RabbitTemplate和SimpleMessageListenerContainer都有一个名为channelTransacted的标志，将该标志设为真，框架使用事务化Channel，在每次操作（发送和接收）后，根据结果自动提交或者撤销事务，并且在撤消后抛出异常。另一个标志是提供一个外部事务（该事务由某个PlatformTransactionManager管理），作为操作的context。在发送和接收消息时，如果channelTransacted标志为真，如果当前已经有一个事务，则消息事务将在当前事务之后提交或撤销，而如果channelTransacted标志为假，则该消息不会产生事务处理，接收到的消息被自动确认（auto-acked）。

channelTransacted标志只能在配置时有效：它只在AMQP组件创建时（通常在应用程序启动时）被使用。外部事务的方式更加灵活，因为系统需要根据线程的当前状态进行处理，但是在实践中通常也是通过配置方式设置的——因为事务通常也是通过配置方式声明的。

在同步接收模式下，外部事务由调用者通过显式或隐式声明提供。下面是一个显式声明的例子，该template已经配置为channelTransacted=true：

@Transactional
public void doSomething() {
String incoming = rabbitTemplate.receiveAndConvert();
// do some more database processing...
String outgoing = processInDatabaseAndExtractReply(incoming);
rabbitTemplate.convertAndSend(outgoing);
}

在上例中，程序接收消息并进行处理，该方法被声明为需要事务支持（@Transactional），因此如果数据库操作失败，接收的消息会被退回到broker，也不会发出回复消息。这个规则适用于在事务性方法链（chain of transactional methods，我理解为方法嵌套）中通过该RabbitTemplate进行的任何操作（除非在例外发生前，Channel中特意提交事务）。

在异步接收模式下，如果需要外部事务，则SimpleMessageListenerContainer在设置listener时必须注入一个PlatformTransactionManager，如下所示：

@Configuration
public class ExampleExternalTransactionAmqpConfiguration {
@Bean
public SimpleMessageListenerContainer messageListenerContainer() {
  SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
  container.setConnectionFactory(rabbitConnectionFactory());
  container.setTransactionManager(transactionManager());
  container.setChannelTransacted(true);
  container.setQueueName("some.queue");
  container.setMessageListener(exampleListener());
  return container;
}
}

在上例中，transaction manager在另一个bean的定义中已经声明为依赖项，channelTransacted设为true，这样当listener处理消息时，如果发生了异常，则事务将会撤销，消息将退回给broker，值得注意的是，如果事务提交失败（例如数据库处理失败，或者遇到连通性问题），则AMQP事务也同样会撤销，消息被退回给broker。这种情况即最优的一阶段提交（Best Efforts 1 Phase Commit），是在可靠消息处理场景下的一个很好的模式。如果channelTransacted设为false（默认情形），尽管listener外部事务同样提供给了listener，但是所有的消息都是自动确认的（auto-acked），即使业务操作失败，消息处理还是成功的。

AMQP事务仅发生于消息或者消息确认（ack）发送到broker时（编者注：我理解AMQP事务对发送消息的情形无法控制，因为一旦发送出去，消息马上就在途了，系统是没有办法帮你撤销的，因此发送消息时程序要做好控制——比如在所有操作都成功时才发送消息），因此当接收消息时，如果事务需要撤销，Spring AMQP所做的事情就是拒绝该消息（很多时候称之为nack，但AMQP规范并没有定义这个名词）。对消息的拒绝处理独立于事务处理，而取决于defaultRequeueRejected属性（默认为真）。关于消息拒绝处理的更多信息，请参考http://www.rabbitmq.com/semantics.html。

使用RabbitTransactionManager

使用RabbitTransactionManager是与外部事务同步执行Rabbit操作的另一种方式。该类实现了PlatformTransactionManager接口，在使用时，必须提供一个Rabbit ConnectionFactory。

注：该种方式并不能提供XA事务支持，例如不能跨消息和数据源共享事务。

另外，应用程序不能直接使用Connection.CreateChannel()方法来创建Channel，必须使用ConnectionFactoryUtils.getTransactionalResourceHolder(ConnectionFactory, boolean)来创建事务性的Rabbit资源。这样，在使用Spring的RabbitTemplate时，它会自动检测到线程敏感的Channel并参与事务处理。

通过Java Configuration你可以创建一个RabbitTransactionManager，如下所示：

@Bean
public RabbitTransactionManager rabbitTransactionManager() {
return new RabbitTransactionManager(connectionFactory);
}

或者使用xml方式

3.11 配置Message Listener Container

配置SimpleMessageContainer时，有相当多的参数与事务和服务有关，下面是一些例子（译者注：由于篇幅太大，只挑选了少数）：

channelTransacted：对于所有接收到的消息，必须在同一个事务中回复；
acknowledgeMode：NONE——消息发送方不会等待消息回复，AUTO——自动发送回复（如果没有抛出异常），MANUAL—

—必须明确调用Channel.basicAck()；
txSize：在自动ack模式下，最多同时处理的消息条数
concurrentConsumers：每个listener初始化时启动的消息接收者个数
receiveTimeout：等待消息进行处理的超时时间
taskExecutor

3.12 Listener并发处理

默认情况下，listener container仅启动一个接收者来处理消息。
你也可以研究上述container参数的效果，很多是用来控制并发处理的参数，最简单的一个参数是concurrentConsumers，该参数的效果是创建固定个数的消息接收者（该参数在1.3.0之后可以动态修改）。

通过maxConcurrentConsumers、consecutiveActiveTrigger、startConsumerMinInterval等参数，还可以根据消息接收

的压力动态地增减消息接收者。

3.13 独占式的消息接收者

1.3版本后，listener container可以配置一个独占式的消息接收者，当然该container的并发数必须为1。这样，当这个接收者工作时（从一个或多个队列处理消息），其他container中的消息接收者必须等待。

3.14 消息接收队列

1.3版本后中对于listener container处理多个消息队列进行了优化。

在1.3版本前，listener container必须至少配置一个消息队列，1.3版本后，消息队列可以被动态地添加和删除，在这种情况下，container将动态地创建或者回收consumers。

1.3版本下其他的优化还包括container允许部分队列失效，并解决了删除队列时会导致整个container停止的问题。

如果container中的队列失效，则consumer会试图每60秒自动创建该失效的队列。因此如果你想彻底删除一个队列，应当在container的配置中移除该队列。

3.15 从错误和故障中恢复

Spring AMQP关键特性之一就是能从协议错误（例如接收到非法数据）和容器错误中进行恢复或自动重连。在前文中我们已经接触到了所有相关的组件，现在我们就recovery的场景进行分析。

最主要的自动重连机制由CachingConnectionFactory提供，当然使用RabbitAdmin的auto-declaration特性也可以提供自动重连。如果你关注可靠的消息传递，你可能需要在RabbitTemplate和SimpleMessageListenerContainer中将channelTransacted设为真以及在SimpleMessageListenerContainer对消息进行回复（或自动回复）。

Exchanges、Queues和Bindings的自动创建

RabbitAdmin可以自动创建Exchanges、Queues和Bindings，它是通过ConnectionListener来侦听是否有消息送达或者需要发送。有了这一套机制，在连接失效时（比如容器停止响应、网络故障等情况下），在下次需要时这些资源能够自动创建。

自动创建的队列名必须是固定的名字，或者匿名队列。

要使用自动创建，CachingConnectionFactory的Cache Mode必须是Channel。

同步操作中的故障恢复

如果以同步的方式使用RabbitTemplate时，遇到了丢失连接的情况，Spring AMQP会抛出AmqpException（通常是AmqpIOException，当然不排除其他的类型），我们希望应用程序自己来处理这样的异常。最简单的情况是，如果你觉得不是通信以外的其他原因，那么可以进行重试。

你可以自己进行重试，也可以使用Spring Retry框架。Spring Retry提供了一系列AOP拦截器可以灵活设置重试的次数、间隔时间等参数。Spring AMQP为创建和使用Spring Retry的拦截器提供了一些方便的组件，并提供了回调的界面使你可以自定义恢复的实现过程，参见StatefulRetryOperationsInterceptor和StatelessRetryOperationsInterceptor。

当AMQP操作不需要事务支持，或者事务在每个retry回调中启动时，我们倾向于使用无状态的retry，这样配置起来相对简单一些。但是当retry处理需要回退时，则我们倾向于时使用有状态的retry。

1.3版本后，有一个方便的API可以使用java configuration方式创建拦截器，例如：

@Bean
public StatefulRetryOperationsInterceptor interceptor() {
return RetryInterceptorBuilder.stateful()
.maxAttempts(5)
.backOffOptions(1000, 2.0, 10000) // initialInterval, multiplier, maxInterval
.build();
}

通过这种方式只能配置一部分重试的参数，如果要配置更多的参数，则可以通过RetryTemplate来进行。

Message Listener和异步消息接收

如果MessageListener遇到了业务侧的异常，则它会捕获该异常，并且继续处理下一条消息，如果遇到了连接失效的异常，则该Listener的consumers将被cancel和restart。SimpleMessageListenerContainer完美地处理这一切，它只是通过log告知listener重启了。事实上Listener会不停地循环试图创建consumer，直到consumer的表现非常糟糕，不得不放弃。

这样做的另一个效果是，在message container启动时，如果broker并没有运行，则它会不停地尝试直到能够建立与broker的连接。

对于业务侧的异常，可能需要更多的考虑和自定义的行为，尤其是使用了事务和container的ack。在2.8.x版本以前，RabbitMQ并没有定义Dead Letter，因此由于业务侧原因拒绝或者roll back的消息可能被永无止境地重新提交。为了限制重新提交的次数，其中之一的选择是使用StatefulRetryOperationsInterceptor。另一个方法是将容器的rejectRequeued属性设为false，这样的效果将使所有失败的消息都被丢弃。

自RabbitMQ 2.8.x版本以后，我们可以通过Dead Letter Exchange来处理那些失败的消息。或者，你也可以抛出AmqpRejectAndDontRequeueException异常。

通常情况下，这些措施被组合使用，例如通过StatefulRetryOperationsInterceptor来限制重试次数，在重试次数到达后使用MessageRecover来进行恢复。默认情况下，MessageRecover只是消费了该消息并发出一条WARNING。不管那个情况，该失败的消息都不会发送到Dead Letter Exchange。

1.3版本后，Spring AMQP提供了RepublishMessageRecoverer，允许在重试次数耗尽后重新发布该消息：

@Bean
RetryOperationsInterceptor interceptor() {
return RetryInterceptorBuilder.stateless()
  .withMaxAttempts(5)
  .setRecoverer(new RepublishMessageRecoverer(amqpTemplate(), "bar", "baz"))
  .build();
}

区分异常类型的重试

Spring Retry有一个很好的机制来区别哪些异常需要进行retry。默认情况下是所有异常都会retry。用户异常被封装在ListenerExecutionFailedException中，用来检查和分类器（Classifier）的匹配关系。默认的Classifier只检查顶层的异常。

从Spring Retry 1.0.3版本以后，BinaryExceptionClassifier提供了一个traverseCauses属性（默认为假），如果设置为真，则该Classifier会分析嵌套的异常。

如果要使用这种Classifier，创建一个SimpleRetryPolicy，并在构造函数中提供最大重试次数，用来匹配的Exception的一个Map，以及traverseCauses的布尔值，并且将该policy注入到RetryTemplate中。

3.16 调试

Spring AMQP提供了大量的日志信息，尤其是DEBUG级别的信息。

如果你想要监视应用程序和broker之间的基于AMQP协议的数据，你可以使用WireShark工具，该工具是一个插件，可以对该协议进行解码。另外RabbitMQ也提供了一个非常有用的类Tracer。该类作为main运行时，默认情况下会侦听localhost的5673和5672端口。运行它，并且将你的connection factory配置到5673端口，它将在console中显示解码后的AMQP协议内容。参见Tracer。