3. Reference

参考文档的这一部分详细介绍了组成 Spring AMQP 的各种组件。 main chapter涵盖了开发 AMQP 应用程序的核心类。本部分还包括有关sample applications的章节。

3.1 使用 Spring AMQP

在本章中，我们将探讨作为使用 Spring AMQP 开发应用程序必不可少的组成部分的接口和类。

3.1.1 AMQP 抽象

Introduction

Spring AMQP 由几个模块组成，每个模块在发行版中均由 JAR 表示。这些模块是：spring-amqp 和 spring-rabbit。 * spring-amqp 模块包含org.springframework.amqp.core软件包。在该软件包中，您将找到代表核心 AMQP“模型”的类。我们的意图是提供不依赖任何特定 AMQP 代理实现或 Client 端库的通用抽象。最终用户代码将只能在抽象层上开发，因此在各个供应商的实现中将更具可移植性。这些抽象然后由特定于代理的模块(例如 spring-rabbit *)实现。当前只有 RabbitMQ 实施；但是，除了 RabbitMQ 之外，还使用 Apache Qpid 在.NET 中对抽象进行了验证。由于 AMQP 原则上在协议级别运行，因此 RabbitMQClient 端可以与支持相同协议版本的任何代理一起使用，但是我们目前不测试任何其他代理。

此处的概述假定您已经熟悉 AMQP 规范的基础知识。如果不是，请查看第 5 章，其他资源中列出的资源。

Message

0-9-1 AMQP 规范未定义消息类或接口。而是在执行诸如basicPublish()之类的操作时，将内容作为字节数组参数传递，并将其他属性作为单独的参数传递。 Spring AMQP 将 Message 类定义为更通用的 AMQP 域模型表示形式的一部分。 Message 类的目的是将正文和属性简单地封装在一个实例中，从而使 API 变得更简单。 Message 类的定义非常简单。

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() {
        return this.body;
    }

    public MessageProperties getMessageProperties() {
        return this.messageProperties;
    }
}

MessageProperties接口定义了几个常用属性，例如* messageId ， timestamp ， contentType 等。可以通过调用setHeader(String key, Object value)方法，使用用户定义的 header *扩展这些属性。

Exchange

Exchange接口代表一个 AMQP 交换，这是消息生产者发送到的对象。代理的虚拟主机中的每个 Exchange 将具有唯一的名称以及一些其他属性：

public interface Exchange {

    String getName();

    String getExchangeType();

    boolean isDurable();

    boolean isAutoDelete();

    Map<String, Object> getArguments();

}

如您所见，Exchange 还具有一个由ExchangeTypes中定义的常量表示的* type 。基本类型是：Direct，Topic，Fanout和Headers。在核心软件包中，您将找到每种类型的Exchange接口的实现。这些 Exchange 类型在处理队列绑定方面的行为各不相同。例如，直接交换允许队列由固定的路由键(通常是队列的名称)绑定。主题交换支持具有路由模式的绑定，这些路由模式可能分别包括**和*#通配符，分别用于 exactly-one 和 zero-or-more *。扇出交换发布到与其绑定的所有队列，而无需考虑任何路由键。有关这些和其他 Exchange 类型的更多信息，请查看第 5 章，其他资源。

Queue

Queue类代表消息使用者从中接收消息的组件。像各种 Exchange 类一样，我们的实现旨在作为这种核心 AMQP 类型的抽象表示。

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

}

请注意，构造函数采用队列名称。取决于实现方式，Management 模板可以提供用于生成唯一命名的队列的方法。这样的队列可用作“答复”地址或其他“临时”情况。因此，自动生成的 Queue 的* exclusive 和 autoDelete 属性都将设置为 true *。

Binding

假定生产者发送到 Exchange，而使用者从队列接收，则将队列连接到 Exchange 的绑定对于通过消息连接这些生产者和使用者至关重要。在 Spring AMQP 中，我们定义了一个Binding类来表示这些连接。让我们回顾一下将队列绑定到 Exchange 的基本选项。

您可以使用固定的路由密钥将队列绑定到 DirectExchange。

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

您可以使用路由模式将队列绑定到 TopicExchange。

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

您可以使用路由键将队列绑定到 FanoutExchange。

new Binding(someQueue, someFanoutExchange)

我们还提供BindingBuilder以促进“Fluent 的 API”样式。

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

就其本身而言，Binding 类的实例只是保存有关连接的数据。换句话说，它不是“活动”组件。但是，正如您稍后将在第 3.1.11 节，“配置代理”中看到的那样，AmqpAdmin类可以使用绑定实例来实际触发代理上的绑定操作。另外，正如您将在同一部分中看到的那样，可以在@Configuration类中使用 Spring 的@Bean -style 定义 Binding 实例。还有一个方便的 Base Class，它进一步简化了用于生成与 AMQP 相关的 bean 定义的方法，并识别队列，交换和绑定，以便在应用程序启动时将它们全部在 AMQP 代理上声明。

AmqpTemplate也在核心包中定义。作为实际 AMQP 消息传递涉及的主要组件之一，将在其自己的部分中详细讨论(请参阅第 3.1.4 节“ AmqpTemplate”)。

3.1.2 连接和资源 Management

Introduction

尽管我们在上一节中描述的 AMQP 模型是通用的，并且适用于所有实现，但是当我们进行资源 Management 时，详细信息特定于代理实现。因此，在本节中，我们将集中讨论仅存在于“ spring-rabbit”模块中的代码，因为在这一点上，RabbitMQ 是唯一受支持的实现。

ConnectionFactory接口是用于 Management 与 RabbitMQ 代理的连接的中央组件。 ConnectionFactory实现的职责是提供org.springframework.amqp.rabbit.connection.Connection的实例，该实例是com.rabbitmq.client.Connection的包装器。我们提供的唯一具体实现是CachingConnectionFactory，默认情况下，它构建一个可以由应用程序共享的单个连接代理。可以共享连接，因为与 AMQP 进行消息传递的“工作单元”实际上是一个“通道”(在某些方面，这类似于 JMS 中的 Connection 和 Session 之间的关系)。可以想象，连接实例提供了createChannel方法。 CachingConnectionFactory实现支持这些通道的缓存，并且根据它们是否是事务性的，为通道维护单独的缓存。创建CachingConnectionFactory的实例时，可以通过构造函数提供* hostname 。还应该提供 username 和 password *属性。如果要配置通道缓存的大小(默认值为 25)，则也可以在此处调用setChannelCacheSize()方法。

从* version 1.3 *开始，可以将CachingConnectionFactory配置为缓存连接以及通道。在这种情况下，每次对createConnection()的调用都会创建一个新连接(或从缓存中检索一个空闲的连接)。关闭连接会将其返回到缓存(如果尚未达到缓存大小)。在此类连接上创建的通道也将被缓存。在某些环境中，使用单独的连接可能很有用，例如从 HA 群集中使用负载，并与负载均衡器一起连接到不同的群集成员。将cacheMode设置为CacheMode.CONNECTION。

从*版本 1.5.5 *开始，提供了一个新属性connectionLimit。设置此选项后，它将限制允许的连接总数。设置后，如果达到限制，则使用channelCheckoutTimeLimitawait 连接变为空闲。如果超过时间，则抛出AmqpTimeoutException。

重要的是要了解，缓存大小(默认情况下)不是限制，而仅仅是可以缓存的通道数。高速缓存大小为 10，实际上可以使用任何数量的通道。如果使用了 10 个以上的通道并将它们全部返回到高速缓存，则 10 个通道将进入高速缓存；其余的将在物理上关闭。

从* version 1.6 *开始，默认的通道缓存大小已从 1 增加到 25.在大容量，多线程的环境中，小的缓存意味着将以较高的速率创建和关闭通道。增加默认缓存大小将避免这种开销。您应该通过 RabbitMQ Admin UI 监视正在使用的通道，如果看到许多正在创建和关闭的通道，请考虑进一步增加缓存大小。缓存将仅按需增长(以适应应用程序的并发要求)，因此此更改不会影响现有的小批量应用程序。

从*版本 1.4.2 *开始，CachingConnectionFactory具有属性channelCheckoutTimeout。当此属性大于零时，channelCacheSize成为可在连接上创建的通道数的限制。如果达到限制，则调用线程将阻塞，直到某个通道可用或达到此超时为止，在这种情况下将引发AmqpTimeoutException。

CachingConnectionFactory connectionFactory = new CachingConnectionFactory("somehost");
connectionFactory.setUsername("guest");
connectionFactory.setPassword("guest");

Connection connection = connectionFactory.createConnection();

使用 XML 时，配置可能如下所示：

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

可以使用 Rabbit 名称空间快速便捷地创建ConnectionFactory：

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

在大多数情况下，这是可取的，因为框架可以为您选择最佳的默认值。创建的实例将是CachingConnectionFactory。请记住，通道的默认缓存大小为 25.如果要缓存更多通道，请通过* channelCacheSize *属性设置一个较大的值。在 XML 中，它看起来像这样：

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

默认的缓存模式是 CHANNEL，但是您可以将其配置为缓存连接。在这种情况下，我们使用connection-cache-size：

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

主机和端口属性可以使用名称空间提供

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

或者，如果在群集环境中运行，请使用地址属性。

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

这是一个自定义线程工厂的示例，该线程工厂在线程名称前加上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>

Naming Connections

从* 1.7 版本*开始，提供了ConnectionNameStrategy以便注入AbstractionConnectionFactory。生成的名称用于目标 RabbitMQ 连接的特定于应用程序的标识。如果 RabbitMQ 服务器支持，则连接名称将显示在 ManagementUI 中。此值不必是唯一的，也不能用作连接标识符，例如在 HTTP API 请求中。该值应该是人类可读的，并且是connection_name键下ClientProperties的一部分。可以使用一个简单的 Lambda：

connectionFactory.setConnectionNameStrategy(connectionFactory -> "MY_CONNECTION");

ConnectionFactory参数可用于通过某些逻辑来区分目标连接名称。默认情况下，AbstractConnectionFactory的beanName，代表对象的十六进制字符串和内部计数器用于生成connection_name。 <rabbit:connection-factory>名称空间组件也随connection-name-strategy属性一起提供。

提供了实现SimplePropertyValueConnectionNameStrategy，该实现将连接名称设置为应用程序属性。将其声明为@Bean并将其注入连接工厂：

@Bean
public ConnectionNameStrategy cns() {
    return new SimplePropertyValueConnectionNameStrategy("spring.application.name");
}

@Bean
public ConnectionFactory rabbitConnectionFactory(ConnectionNameStrategy cns) {
    CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
    ...
    connectionFactory.setConnectionNameStrategy(cns);
    return connectionFactory;
}

该属性必须存在于应用程序上下文的Environment中。

阻止的连接和资源限制

根据Memory Alarm，该连接可能被阻止与 Broker 进行交互。从* version 2.0 *开始，可以为_提供com.rabbitmq.client.BlockedListener，以通知连接被阻止和未被阻止的事件。此外，AbstractConnectionFactory通过其内部BlockedListener实现分别发出ConnectionBlockedEvent和ConnectionUnblockedEvent。这些使您能够提供应用程序逻辑，以对代理程序上的问题做出适当反应，并采取一些纠正措施，例如。

从版本 2.0.2 *开始，RabbitTemplate具有配置选项，以自动使用第二个连接工厂，除非正在使用事务。有关更多信息，请参见“使用单独的连接”部分。发布者连接的ConnectionNameStrategy与主要策略相同，在调用方法的结果后附加.publisher。

从* version 1.7.7 *开始，提供了一个AmqpResourceNotAvailableException，例如当SimpleConnection.createChannel()无法创建Channel时抛出该AmqpResourceNotAvailableException，因为达到了channelMax的限制并且缓存中没有可用的通道。可以在RetryPolicy中使用此异常，以在某些回退之后恢复操作。

配置基础 Client 端连接工厂

CachingConnectionFactory使用 RabbitClient 端ConnectionFactory的实例；在CachingConnectionFactory上设置等效属性时，会传递许多配置属性(例如host, port, userName, password, requestedHeartBeat, connectionTimeout)。要设置其他属性(例如clientProperties)，请定义 Rabbit 工厂的实例，并使用CachingConnectionFactory的适当构造函数为其提供引用。如上所述使用命名空间时，请在connection-factory属性中提供对已配置工厂的引用。为方便起见，提供了工厂 bean 来帮助在 Spring 应用程序上下文中配置连接工厂，如下一节所述。

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

RabbitConnectionFactoryBean 和配置 SSL

从* version 1.4 *开始，提供了一个方便的RabbitConnectionFactoryBean，以使用依赖项注入在基础 Client 端连接工厂上方便地配置 SSL 属性。其他设置员仅委托给基础工厂。以前，您必须以编程方式配置 SSL 选项。

<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.amqp.rabbit.connection.RabbitConnectionFactoryBean">
    <property name="useSSL" value="true" />
    <property name="sslPropertiesLocation" value="file:/secrets/rabbitSSL.properties"/>
</bean>

有关配置 SSL 的信息，请参考RabbitMQ Documentation。省略keyStore和trustStore配置以通过 SSL 进行连接而无需证书验证。密钥和信任库配置可以如下提供：

sslPropertiesLocation属性是 Spring Resource，它指向包含以下键的属性文件：

keyStore=file:/secret/keycert.p12
trustStore=file:/secret/trustStore
keyStore.passPhrase=secret
trustStore.passPhrase=secret

keyStore和truststore是指向 Store 的 Spring Resources。通常，此属性文件将由 os 保护，并且应用程序具有读取访问权限。

从 Spring AMQP *版本 1.5 *开始，可以直接在工厂 bean 上设置这些属性。如果同时提供了离散属性和sslPropertiesLocation，则后者中的属性将覆盖离散值。

路由连接工厂

从* version 1.3 *开始，引入了AbstractRoutingConnectionFactory。这提供了一种机制，可在运行时为多个ConnectionFactories配置 Map 并由某个lookupKey确定目标ConnectionFactory。通常，实现检查线程绑定上下文。为了方便起见，Spring AMQP 提供了SimpleRoutingConnectionFactory，它从SimpleResourceHolder获取当前线程绑定的lookupKey：

<bean id="connectionFactory"
      class="org.springframework.amqp.rabbit.connection.SimpleRoutingConnectionFactory">
	<property name="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());
    }

}

使用后解除绑定资源很重要。有关更多信息，请参见AbstractRoutingConnectionFactory的 JavaDocs。

从*版本 1.4 *开始，RabbitTemplate支持 SpEL sendConnectionFactorySelectorExpression和receiveConnectionFactorySelectorExpression属性，这些属性在每个 AMQP 协议交互操作(send，sendAndReceive，receive或receiveAndReply)上进行评估，对于提供的AbstractRoutingConnectionFactory解析为lookupKey值。表达式中可以使用 Bean 引用，例如"@vHostResolver.getVHost(#root)"。对于send操作，要发送的消息是根评估对象。对于receive操作，queueName 是根评估对象。

**路由算法为：如果 selectors 表达式为null，或者计算为null，或者提供的ConnectionFactory不是AbstractRoutingConnectionFactory的实例，则所有操作都像以前一样，取决于提供的ConnectionFactory实现。如果评估结果不是null，但是没有针对该lookupKey的目标ConnectionFactory，并且AbstractRoutingConnectionFactory配置为lenientFallback = true，则会发生相同的情况。当然，在AbstractRoutingConnectionFactory的情况下，它会回退到基于determineCurrentLookupKey()的routing实现。但是，如果lenientFallback = false，则抛出IllegalStateException。

命名空间支持还在<rabbit:template>组件上提供了send-connection-factory-selector-expression和receive-connection-factory-selector-expression属性。

同样从* version 1.4 *开始，您可以在侦听器容器中配置路由连接工厂。在这种情况下，队列名称列表将用作查找关键字。例如，如果您使用setQueueNames("foo", "bar")配置容器，则查找键将为"[foo,bar]"(无空格)。

从*版本 1.6.9 *开始，您可以使用侦听器容器上的setLookupKeyQualifier向查找键添加限定符。例如，这将允许侦听具有相同名称但在不同虚拟主机中的队列(每个虚拟主机中都有一个连接工厂)。

例如，在使用查找键限定符foo和侦听队列bar的容器的情况下，用于注册目标连接工厂的查找键将是foo[bar]。

队列相似性和 LocalizedQueueConnectionFactory

在群集中使用 HA 队列时，为了获得最佳性能，可能需要连接到主队列所在的物理代理。 CachingConnectionFactory可以配置多个代理地址；这是为了进行故障转移，Client 端将尝试按 Sequences 连接。 LocalizedQueueConnectionFactory使用 Management 插件提供的 REST API 来确定要控制队列的节点。然后，它创建(或从缓存中检索)CachingConnectionFactory，该CachingConnectionFactory将仅连接到该节点。如果连接失败，那么将确定新的主节点，并且使用者将连接到该主节点。 LocalizedQueueConnectionFactory配置有默认的连接工厂，以防无法确定队列的物理位置，在这种情况下，它将正常连接到群集。

LocalizedQueueConnectionFactory是RoutingConnectionFactory，而SimpleMessageListenerContainer使用队列名称作为查找关键字，如上面名为“路由连接工厂”的部分所述。

这是一个示例配置，使用 Spring Boot 的 RabbitProperties 配置工厂：

@Autowired
private RabbitProperties props;

private final String[] adminUris = { "http://host1:15672", "http://host2:15672" };

private final String[] nodes = { "[emailprotected]", "[emailprotected]" };

@Bean
public ConnectionFactory defaultConnectionFactory() {
    CachingConnectionFactory cf = new CachingConnectionFactory();
    cf.setAddresses(this.props.getAddresses());
    cf.setUsername(this.props.getUsername());
    cf.setPassword(this.props.getPassword());
    cf.setVirtualHost(this.props.getVirtualHost());
    return cf;
}

@Bean
public ConnectionFactory queueAffinityCF(
        @Qualifier("defaultConnectionFactory") ConnectionFactory defaultCF) {
    return new LocalizedQueueConnectionFactory(defaultCF,
            StringUtils.commaDelimitedListToStringArray(this.props.getAddresses()),
            this.adminUris, this.nodes,
            this.props.getVirtualHost(), this.props.getUsername(), this.props.getPassword(),
            false, null);
}

请注意，前三个参数是addresses，adminUris和nodes的数组。这些是适当的，因为当容器尝试连接到队列时，它确定队列在哪个节点上被控制，并连接到同一阵列位置中的地址。

发布者确认并return

通过将CachingConnectionFactory的publisherConfirms和publisherReturns属性分别设置为'true'，可以支持确认和返回的消息。

设置这些选项后，工厂创建的Channel将被包装在PublisherCallbackChannel中，该_用于方便回调。当获得这样的 Channels 时，Client 端可以向Channel注册PublisherCallbackChannel.Listener。 PublisherCallbackChannel实现包含将确认/返回路由到适当的侦听器的逻辑。这些功能将在以下各节中进一步说明。

另请参见称为“范围操作”的部分中的simplePublisherConfirms。

连接和 Channels 监听器

连接工厂支持注册ConnectionListener和ChannelListener实现。这使您可以接收有关连接和通道相关事件的通知。 (构建连接时，RabbitAdmin使用ConnectionListener来执行声明-有关更多信息，请参见称为“交换，队列和绑定的自动声明”的部分)。

ConnectionListener.

@FunctionalInterface
public interface ConnectionListener {

    void onCreate(Connection connection);

    default void onClose(Connection connection) {
    }

    default void onShutDown(ShutdownSignalException signal) {
    }

}

从* version 2.0 *开始，可以为org.springframework.amqp.rabbit.connection.Connection对象提供com.rabbitmq.client.BlockedListener s 来通知连接被阻止和未被阻止的事件。

ChannelListener.

@FunctionalInterface
public interface ChannelListener {

    void onCreate(Channel channel, boolean transactional);

    default void onShutDown(ShutdownSignalException signal) {
    }

}

有关您可能想注册ChannelListener的一种情况，请参见“发布是异步的-如何检测成功和失败”部分。

记录 Channels 关闭事件

在 1.5 版中引入了一种使用户能够控制日志记录级别的机制。

CachingConnectionFactory使用默认策略记录通道关闭，如下所示：

• 正常通道关闭(200 OK)不会被记录。

• 如果通道由于被动队列声明失败而关闭，那么它将在调试级别记录。

• 如果通道由于basic.consume由于特殊的使用者条件而被拒绝而关闭，则它将以 INFO 级别记录。

• 其他所有日志均以 ERROR 级别记录。

若要修改此行为，请在其closeExceptionLogger属性中将自定义ConditionalExceptionLogger注入到CachingConnectionFactory中。

另请参见“Consumer 事件”部分。

运行时缓存属性

从* version 1.6 *开始，CachingConnectionFactory现在通过getCacheProperties()方法提供了缓存统计信息。这些统计信息可用于调整缓存以在 Producing 对其进行优化。例如，高水位标记可用于确定是否应增加缓存大小。如果等于缓存大小，则可能要考虑进一步增加。

表 3.1. CacheMode.CHANNEL 的缓存属性

Property	Meaning
connectionName	ConnectionNameStrategy生成的连接的名称。
channelCacheSize	当前配置的允许空闲的最大通道数。
localPort	连接的本地端口(如果有)。这可用于与 RabbitMQ Admin UI 上的连接/通道关联。
idleChannelsTx	当前空闲(缓存)的事务通道的数量。
idleChannelsNotTx	当前空闲(缓存)的非事务通道的数量。
idleChannelsTxHighWater	已同时空闲(缓存)的最大事务通道数。
idleChannelsNotTxHighWater	非事务通道的最大数量已被同时空闲(缓存)。

表 3.2. CacheMode.CONNECTION 的缓存属性

Property	Meaning
connectionName:<localPort>	ConnectionNameStrategy生成的连接的名称。
openConnections	表示与代理的连接的连接对象的数量。
channelCacheSize	当前配置的允许空闲的最大通道数。
connectionCacheSize	当前配置的允许空闲的最大连接数。
idleConnections	当前空闲的连接数。
idleConnectionsHighWater	并发空闲的最大连接数。
idleChannelsTx:<localPort>	该连接当前空闲(缓存)的事务通道的数量。属性名称的 localPort 部分可用于与 RabbitMQ Admin UI 上的连接/通道关联。
idleChannelsNotTx:<localPort>	该连接当前空闲(缓存)的非事务通道的数量。属性名称的 localPort 部分可用于与 RabbitMQ Admin UI 上的连接/通道关联。
idleChannelsTxHighWater:
<localPort>	同时空闲(缓存)的最大事务通道数。属性名称的 localPort 部分可用于与 RabbitMQ Admin UI 上的连接/通道关联。
idleChannelsNotTxHighWater: <localPort>	已同时空闲(缓存)的非事务通道的最大数量。属性名称的 localPort 部分可用于与 RabbitMQ Admin UI 上的连接/通道关联。

cacheMode属性(还包括CHANNEL或CONNECTION)。

图 3.1. JVisualVM 示例

RabbitMQ 自动连接/拓扑恢复

从 Spring AMQP 的第一个版本开始，该框架在代理发生故障的情况下提供了自己的连接和通道恢复。另外，如第 3.1.11 节，“配置代理”中所述，RabbitAdmin将在重新构建连接时重新声明任何基础结构 Bean(队列等)。因此，它不依赖amqp-client库现在提供的Auto Recovery。 Spring AMQP 现在使用amqp-client的4.0.x版本，默认情况下启用了自动恢复。如果愿意，Spring AMQP 仍可以使用其自己的恢复机制，在 Client 端中将其禁用(通过将基础RabbitMQ connectionFactory设置为false的automaticRecoveryEnabled属性)。但是，该框架与启用的自动恢复完全兼容。这意味着您在代码中创建的所有使用者(可能通过RabbitTemplate.execute())都可以自动恢复。

3.1.3 添加自定义 Client 端连接属性

CachingConnectionFactory现在允许您访问基础连接工厂，以允许例如设置自定义 Client 端属性：

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

查看连接时，这些属性会显示在 RabbitMQManagement 员 UI 中。

3.1.4 AmqpTemplate

Introduction

与 Spring 框架和相关项目提供的许多其他高级抽象一样，Spring AMQP 提供了一个“模板”，该模板起着核心作用。定义主要操作的接口称为AmqpTemplate。这些操作涵盖了发送和接收消息的一般行为。换句话说，它们不是任何实现所独有的，因此名称中为“ AMQP”。另一方面，该接口的实现与 AMQP 协议的实现绑定在一起。与 JMS(本身是接口级 API)不同，AMQP 是线级协议。该协议的实现提供了自己的 Client 端库，因此模板接口的每种实现都将取决于特定的 Client 端库。当前，只有一个实现：RabbitTemplate。在下面的示例中，您经常会看到“ AmqpTemplate”的用法，但是当您查看配置示例或实例化模板和/或调用设置器的任何代码摘录时，您将看到实现类型(例如“ RabbitTemplate”)。

如上所述，AmqpTemplate接口定义了用于发送和接收消息的所有基本操作。在接下来的两个部分中，我们将分别探讨消息发送和接收。

另请参见名为“ AsyncRabbitTemplate”的部分。

添加重试功能

从*版本 1.3 *开始，您现在可以将RabbitTemplate配置为使用RetryTemplate来帮助处理代理连接性问题。有关完整信息，请参见spring-retry项目。以下仅是一个使用指数回退策略和默认值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;
}

从* version 1.4 *开始，除了retryTemplate属性之外，RabbitTemplate还支持recoveryCallback选项。用作RetryTemplate.execute(RetryCallback<T, E> retryCallback, RecoveryCallback<T> 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>() {

        @Override
        public Object recover(RetryContext context) throws Exception {
            Object message = context.getAttribute("message");
            Throwable t = context.getLastThrowable();
            // Do something with message
            return null;
        }
    });
}

在这种情况下，您将 不 将RetryTemplate注入RabbitTemplate。

发布是异步的-如何检测成功和失败

发布消息是一种异步机制，默认情况下，RabbitMQ 会丢弃无法路由的消息。为了成功发布，您可以收到异步确认，如下称为“发布者确认并返回”的部分所述。让我们考虑两种故障情况：

• 发布到 Transaction 所，但没有匹配的目标队列

• 发布到不存在的交换

如以下称为“发布者确认并返回”的部分所述，第一种情况由发布商return涵盖。

在第二种情况下，该消息被丢弃，不生成任何返回。底层通道被异常关闭。默认情况下，会记录此异常，但是您可以向CachingConnectionFactory注册ChannelListener以获得此类事件的通知：

this.connectionFactory.addConnectionListener(new ConnectionListener() {

    @Override
    public void onCreate(Connection connection) {
    }

    @Override
    public void onShutDown(ShutdownSignalException signal) {
        ...
    }

});

您可以检查 signal 的reason属性以确定发生的问题。

要检测发送线程上的异常，可以在RabbitTemplate上setChannelTransacted(true)，并将在txCommit()上检测到异常。但是，“Transaction 会严重影响性能” **，因此在仅针对一个使用案例启用 Transaction 之前，请仔细考虑这一点。

发布者确认并return

AmqpTemplate的RabbitTemplate实现支持发布者确认和return。

对于返回的消息，模板的mandatory属性必须设置为true，或者对于特定消息，mandatory-expression必须计算为true。此功能需要CachingConnectionFactory的publisherReturns属性设置为 true(请参见称为“发布者确认并返回”的部分)。通过调用setReturnCallback(ReturnCallback callback)注册RabbitTemplate.ReturnCallback，将返回值发送给 Client 端。回调必须实现此方法：

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

每个RabbitTemplate仅支持一个ReturnCallback。另请参见称为“答复超时”的部分。

对于发布者确认(又称发布者确认)，模板需要一个CachingConnectionFactory，其publisherConfirms属性设置为 true。通过调用setConfirmCallback(ConfirmCallback callback)注册RabbitTemplate.ConfirmCallback来将确认发送到 Client 端。回调必须实现此方法：

void confirm(CorrelationData correlationData, boolean ack, String cause);

CorrelationData是 Client 端在发送原始消息时提供的对象。 ack对ack为 true，对nack为 false。对于nack，如果生成nack时原因可用，则原因可能包含 nack 的原因。一个示例是将消息发送到不存在的交换机时。在这种情况下，broker 关闭 Channel； cause中包含了关闭的原因。 cause已添加到*版本 1.4 *中。

RabbitTemplate仅支持一个ConfirmCallback。

从版本 2.1 开始，CorrelationData对象具有ListenableFuture，可用于获取结果，而不是在模板上使用ConfirmCallback。

CorrelationData cd1 = new CorrelationData();
this.templateWithConfirmsEnabled.convertAndSend("exchange", queue.getName(), "foo", cd1);
assertTrue(cd1.getFuture().get(10, TimeUnit.SECONDS).isAck());

由于它是ListenableFuture<Confirm>，因此您可以在准备好get()的结果，或为异步回调添加侦听器。 Confirm对象是一个简单的 Bean，具有 2 个属性ack和reason(用于 nacks)。broker 产生的 na 俩并未填充原因；它填充了框架产生的漏洞(例如，在漏洞未解决时关闭连接)。

此外，当启用了确认和返回功能时，CorrelationData将填充返回的消息。可以肯定的是，在使用 ack 设置将来之前会发生这种情况。

另请参见称为“范围操作”的部分，以 await 发布者确认的更简单机制。

Scoped Operations

通常，在使用模板时，会从缓存中签出Channel(或创建一个Channel)，并将其用于操作，然后将其返回到缓存中以进行重用。在多线程环境中，不能保证下一个操作将使用相同的通道。但是，有时您可能希望更好地控制通道的使用，并确保在同一通道上执行全部操作。

从* version 2.0 *版本开始，提供了一个新方法invoke，其中一个OperationsCallback。在回调范围内以及提供的RabbitOperations参数上执行的任何操作都将使用相同的专用Channel，该专用变量Channel将在最后关闭(不返回到缓存)。如果通道是PublisherCallbackChannel，则在收到所有确认后，它将返回到缓存(请参见上面的称为“发布者确认并返回”的部分)。

@FunctionalInterface
public interface OperationsCallback<T> {

    T doInRabbit(RabbitOperations operations);

}

您可能需要这样做的一个示例是，如果您希望在基础Channel上使用waitForConfirms()方法。以前未使用 Spring API 公开此方法，因为通常如上所述对通道进行缓存和共享。 RabbitTemplate现在提供waitForConfirms(long timeout)和waitForConfirmsOrDie(long timeout)，它们委派给OperationsCallback范围内使用的专用通道。由于明显的原因，不能在该范围之外使用这些方法。

请注意，在其他位置(请参见称为“发布者确认并返回”的部分)提供了更高级别的抽象，该抽象使您可以将确认与请求相关联。如果您只想 awaitbroker 确认交货，则可以使用以下简单技术：

Collection<?> messages = getMessagesToSend();
Boolean result = this.template.invoke(t -> {
    messages.forEach(m -> t.convertAndSend(ROUTE, m));
    t.waitForConfirmsOrDie(10_000);
    return true;
});

如果您希望在OperationsCallback范围内在同一通道上调用RabbitAdmin操作，则必须使用与invoke操作相同的RabbitTemplate构造 Management 员。

当以这种方式使用确认时，实际上并不需要构建用于将确认与请求相关联的许多基础结构。从* version 2.1 *开始，连接工厂支持新属性simplePublisherConfirms。如果为true，则可以避免基础结构，并且确认处理将更加有效。

此外，RabbitTemplate在发送的消息MessageProperties中设置属性publisherSequenceNumber。如果您希望检查(或记录等)特定的确认，可以使用重载的invoke方法进行检查：

public <T> T invoke(OperationsCallback<T> action, com.rabbitmq.client.ConfirmCallback acks,
        com.rabbitmq.client.ConfirmCallback nacks);

Example:

Collection<?> messages = getMessagesToSend();
Boolean result = this.template.invoke(t -> {
    messages.forEach(m -> t.convertAndSend(ROUTE, m));
    t.waitForConfirmsOrDie(10_000);
    return true;
}, (tag, multiple) -> {
        log.info("Ack: " + tag + ":" + multiple);
}, (tag, multiple) -> {
        log.info("Nack: " + tag + ":" + multiple);
}));

Messaging integration

从* version 1.4 * RabbitMessagingTemplate开始，它构建在RabbitTemplate的基础上，提供了与 Spring Framework 消息传递抽象(即org.springframework.messaging.Message)的集成。这使您可以使用spring-messaging Message<?>抽象来发送和接收消息。其他 Spring 项目(例如 Spring Integration 和 Spring 的 STOMP 支持)使用了这种抽象。涉及两个消息转换器。一种在 Spring 消息Message<?>和 Spring AMQP 的Message抽象之间进行转换，另一种在 Spring AMQP 的Message抽象与基础 RabbitMQClient 端库所需的格式之间进行转换。默认情况下，消息有效负载由提供的RabbitTemplate的消息转换器转换。或者，您可以将自定义MessagingMessageConverter注入其他一些有效负载转换器：

MessagingMessageConverter amqpMessageConverter = new MessagingMessageConverter();
amqpMessageConverter.setPayloadConverter(myPayloadConverter);
rabbitMessagingTemplate.setAmqpMessageConverter(amqpMessageConverter);

已验证的用户 ID

从* version 1.6 *开始，模板现在支持user-id-expression(使用 Java 配置时为userIdExpression)。如果发送了一条消息，则在评估此表达式后将设置用户 id 属性(如果尚未设置)。评估的根对象是要发送的消息。

Examples:

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

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

第一个示例是一个 Literals 表达式；第二个从应用程序上下文中的连接工厂 Bean 获取username属性。

使用单独的连接

从* version 2.0.2 *开始，在可能的情况下，将usePublisherConnection属性设置为true以使用与侦听器容器使用的连接不同的连接。这是为了避免在生产者由于任何原因被阻止时阻止 Consumer。 CachingConnectionFactory现在为此维护了第二个内部连接工厂。如果 Rabbit 模板在由侦听器容器启动的事务中运行，则无论此设置如何，都将使用容器的通道。

3.1.5 发送消息

Introduction

发送邮件时，可以使用以下任何一种方法：

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 交换名称以及路由密钥。最后一个参数是负责实际创建 Message 实例的回调。使用此方法发送消息的示例可能如下所示：

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

如果您计划大部分或所有时间使用模板实例发送到同一交换，则可以在模板本身上设置“交换”属性。在这种情况下，可以替代使用上面列出的第二种方法。以下示例在功能上等同于上一个示例：

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));

考虑交换和路由键属性的一种更好的方法是，显式方法参数将始终覆盖模板的默认值。实际上，即使您没有在模板上显式设置这些属性，也始终会存在默认值。在这两种情况下，默认值都是一个空字符串，但这实际上是一个明智的默认值。就路由密钥而言，并非一开始就总是必需的(例如，扇出交换)。此外，队列可以使用空字符串绑定到 Exchange。这些都是合法的方案，它们依赖模板的路由键属性的默认空字符串值。就 Exchange 名称而言，空字符串非常常用，因为 AMQP 规范将“默认 Exchange”定义为没有名称。由于所有队列都使用其名称作为绑定值自动绑定到该默认 Exchange(即直接 Exchange)，因此上述第二种方法可用于通过默认 Exchange 简单地点对点消息传递到任何队列。只需提供队列名称作为“ routingKey”-通过在运行时提供 method 参数即可：

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

或者，如果您希望创建一个模板，该模板将主要或专门用于发布到单个 Queue，则以下内容是完全合理的：

RabbitTemplate template = new RabbitTemplate(); // using default no-name Exchange
template.setRoutingKey("queue.helloWorld"); // but we'll always send to this Queue
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();

or

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();

可以设置MessageProperties上定义的每个属性。其他方法包括setHeader(String key, String value)，removeHeader(String key)，removeHeaders()和copyProperties(MessageProperties properties)。每个属性设置方法都有一个set*IfAbsent()变体。在存在默认初始值的情况下，该方法名为set*IfAbsentOrDefault()。

提供了五个静态方法来创建初始消息构建器：

public static MessageBuilder withBody(byte[] body) (1)

public static MessageBuilder withClonedBody(byte[] body) (2)

public static MessageBuilder withBody(byte[] body, int from, int to) (3)

public static MessageBuilder fromMessage(Message message) (4)

public static MessageBuilder fromClonedMessage(Message message) (5)

• (1) 由构建器创建的消息将具有一个直接引用该参数的正文。
• (2) 由构建器创建的消息将具有一个主体，该主体是一个新数组，在参数中包含字节的副本。
• (3) 由构建器创建的消息将具有一个主体，该主体是一个新数组，其中包含来自参数的字节范围。有关更多详细信息，请参见Arrays.copyOfRange()。
• (4) 由构建器创建的消息将具有直接引用自变量正文的正文。参数的属性将复制到新的MessageProperties对象。
• (5) 由构建器创建的消息将具有一个主体，该主体是一个包含参数主体副本的新数组。参数的属性将复制到新的MessageProperties对象。

public static MessagePropertiesBuilder newInstance() (1)

public static MessagePropertiesBuilder fromProperties(MessageProperties properties) (2)

public static MessagePropertiesBuilder fromClonedProperties(MessageProperties properties) (3)

• (1) 新的消息属性对象使用默认值初始化。
• (2) 构建器将使用提供的属性对象进行初始化，并且build()将返回。
• (3) 参数的属性将复制到新的MessageProperties对象。

使用AmqpTemplate的RabbitTemplate实现，每个send()方法都有一个重载版本，该重载版本带有一个额外的CorrelationData对象。启用发布者确认后，此对象将在第 3.1.4 节“ AmqpTemplate”中描述的回调中返回。这允许发送方将确认(确认或不确认)与发送的消息相关联。

从* version 1.6.7 *开始，引入了CorrelationAwareMessagePostProcessor接口，允许在转换消息后修改相关数据：

Message postProcessMessage(Message message, Correlation correlation);

在 2.0 版中，不推荐使用此接口；该方法已通过默认实现(已委派给postProcessMessage(Message message))移至MessagePostProcessor。

同样从* version 1.6.7 *开始，提供了一个新的回调接口CorrelationDataPostProcessor；在所有MessagePostProcessor之后(在send()方法以及setBeforePublishPostProcessors()中提供的方法中提供)之后调用此方法。实现可以更新或替换send()方法(如果有)中提供的相关数据。 Message和原始CorrelationData(如果有)作为参数提供。

CorrelationData postProcess(Message message, CorrelationData correlationData);

Publisher Returns

当模板的mandatory属性为* true *时，返回的消息由第 3.1.4 节“ AmqpTemplate”中描述的回调提供。

从*版本 1.4 *开始，RabbitTemplate支持 SpEL mandatoryExpression属性，该属性针对每个请求消息进行评估，作为根评估对象，解析为boolean值。表达式中可以使用 Bean 引用，例如"@myBean.isMandatory(#root)"。

RabbitTemplate还可在发送和接收操作中内部使用发布者返回。有关更多信息，请参见称为“答复超时”的部分。

Batching

从*版本 1.4.2 *开始，引入了BatchingRabbitTemplate。这是RabbitTemplate的子类，具有重写的send方法，该方法根据BatchingStrategy批处理消息；仅当批次完成时，消息才会发送到 RabbitMQ。

public interface BatchingStrategy {

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

	Date nextRelease();

	Collection<MessageBatch> releaseBatches();

}

提供了SimpleBatchingStrategy。它支持将消息发送到单个交换/路由键。它具有以下特性：

• batchSize-批量发送之前的邮件数

• bufferLimit-批处理邮件的最大大小；如果超过batchSize，它将抢占batchSize并导致发送部分批次

• timeout-在没有新活动将消息添加到该批处理中之后将发送部分批处理的时间

SimpleBatchingStrategy通过在每个嵌入的消息之前添加 4 字节的二进制长度来格式化批处理。通过将springBatchFormat message 属性设置为lengthHeader4，将其传达给接收系统。

3.1.6 接收消息

Introduction

消息接收总是比发送要复杂一些。接收Message有两种方法。比较简单的选项是使用轮询方法一次轮询单个Message。更复杂但更常见的方法是注册一个将按需异步接收Messages的侦听器。在接下来的两个小节中，我们将介绍每种方法的示例。

Polling Consumer

AmqpTemplate本身可用于轮询消息的接收。默认情况下，如果没有可用消息，则立即返回null；没有阻塞。从* version 1.5 *版本开始，您现在可以设置receiveTimeout(以毫秒为单位)，并且 receive 方法将阻塞长达该时间，await 消息。小于零的值表示无限期阻塞(或至少直到与代理的连接丢失为止)。 *版本 1.6 *引入了receive方法的变体，允许在每次调用中传递超时。

有四种简单的* receive *方法。与发送方的 Exchange 一样，有一种方法要求直接在模板本身上设置默认队列属性，并且有一种方法在运行时接受队列参数。 *版本 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拥有一些方便的方法来接收 POJO 而不是Message实例，并且实现将提供一种自定义MessageConverter的方法，该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;

从* version 2.0 *版本开始，这些方法的变体都带有一个额外的ParameterizedTypeReference参数来转换复杂类型。模板必须配置为SmartMessageConverter；有关更多信息，请参见名为“使用 RabbitTemplate 从消息转换”一节。

与sendAndReceive方法类似，从* version 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的实例(提供原始消息交换 Contract)时，才应用自动消息(请求和答复)转换。

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

Asynchronous Consumer

Message Listener

对于异步消息接收，涉及一个专用组件(不是AmqpTemplate)。该组件是消息消耗回调的容器。我们将在短时间内查看容器及其属性，但是首先应该查看回调，因为这是您的应用程序代码将与消息传递系统集成的地方。从MessageListener接口的实现开始，回调有几个选项：

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

如果您的回调逻辑出于任何原因依赖于 AMQP Channel 实例，则可以改用ChannelAwareMessageListener。看起来很相似，但是有一个额外的参数：

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

MessageListenerAdapter

如果您希望在应用程序逻辑和消息传递 API 之间保持更严格的分隔，则可以依靠框架提供的适配器实现。这通常称为“消息驱动的 POJO”支持。

使用适配器时，仅需要提供对适配器本身应调用的实例的引用。

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

您可以继承适配器的子类并提供getListenerMethodName()的实现，以根据消息动态选择不同的方法。此方法有两个参数originalMessage和extractedMessage，后者是任何转换的结果。默认情况下，配置为SimpleMessageConverter；有关更多信息和其他可用转换器的信息，请参见名为“ SimpleMessageConverter”的部分。

从*版本 1.4.2 *开始，原始消息具有属性consumerQueue和consumerTag，这些属性可用于确定从哪个队列接收消息。

从* version 1.5 *开始，您可以配置使用者队列/标记到方法名称的 Map，以动态选择要调用的方法。如果 Map 中没有条目，我们将使用默认的侦听器方法。默认的侦听器方法(如果未设置)是handleMessage。

从* version **版本开始，提供了一个方便的FunctionalInterface：

@FunctionalInterface
public interface ReplyingMessageListener<T, R> {

	R handleMessage(T t);

}

这有助于使用 Java 8 lamdas 方便地配置适配器：

new MessageListenerAdapter((ReplyingMessageListener<String, String>) data -> {
    ...
    return result;
}));

Container

既然您已经看到了 Message-listening 回调的各种选项，我们就可以将注意力转向容器。基本上，容器处理“主动”职责，以便侦听器回调可以保持被动状态。容器是“生命周期”组件的一个示例。它提供了启动和停止的方法。配置容器时，实际上是在弥合 AMQP 队列和MessageListener实例之间的差距。您必须提供对ConnectionFactory的引用以及侦听器应从中使用 Messages 的队列名称或队列实例。

对于* version 2.0 *之前的版本，只有一个侦听器容器SimpleMessageListenerContainer;现在有了第二个容器DirectMessageListenerContainer。 名为“选择容器”的部分中描述了在选择使用哪种容器时可能会应用的容器和标准之间的差异。

这是使用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>

or

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

将创建一个DirectMessageListenerContainer(请注意type属性-默认为simple)。

或者，您可能更喜欢使用@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);
            }
        };
    }
}

Consumer Priority

从 RabbitMQ 版本 3.2 开始，代理现在支持使用者优先级(请参阅通过 RabbitMQ 使用 Consumer 优先级)。通过在使用者上设置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>

从*版本 1.3 *开始，可以在运行时修改正在侦听容器的队列。参见第 3.1.20 节“侦听器容器队列”。

auto-delete Queues

如果将容器配置为侦听auto-delete队列，或者该队列具有x-expires选项，或者在 Broker 上配置了Time-To-Live策略，则当容器停止(最后一个使用者被取消)时，代理会将队列删除。在* version 1.3 *之前，由于缺少队列，因此无法重新启动容器。 RabbitAdmin仅在连接关闭/打开时自动重新声明队列等，而在容器停止/启动时不会发生这种情况。

从* version 1.3 *开始，容器现在将在启动过程中使用RabbitAdmin重新声明所有丢失的队列。

您还可以结合使用条件声明(称为“有条件的声明”的部分)和auto-startup="false" admin 来将队列声明推迟到容器启动之前。

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

在这种情况下，队列和交换由具有auto-startup="false"的containerAdmin声明，因此在上下文初始化期间不声明元素。同样，由于相同的原因，容器没有启动。稍后启动容器时，它将使用对containerAdmin的引用来声明元素。

Batched Messages

侦听器容器自动分批处理了批处理的邮件(使用springBatchFormat邮件标题)。拒绝批次中的任何消息都将导致整个批次被拒绝。有关批处理的更多信息，请参见称为“批处理”的部分。

Consumer Events

每当侦听器(使用者)遇到某种类型的故障时，容器就会发布应用程序事件。事件ListenerContainerConsumerFailedEvent具有以下属性：

• container-Consumer 遇到问题的侦听器容器。

• reason-失败的 Literals 原因。

• fatal-指示失败是否致命的布尔值；在非致命异常的情况下，容器将根据recoveryInterval或recoveryBackoff(对于SimpleMessageListenerContainer)或monitorInterval(对于DirectMessageListenerContainer)尝试重新启动使用者。

• throwable-被捕获的Throwable。

通过实现ApplicationListener<ListenerContainerConsumerFailedEvent>可以消耗这些事件。

如果使用者由于默认情况下仅使用其队列而失败以及发布事件而失败，则会发出WARN日志。要更改此日志记录行为，请在SimpleMessageListenerContainer的exclusiveConsumerExceptionLogger属性中提供一个自定义ConditionalExceptionLogger。另请参见名为“记录通道关闭事件”的部分。

致命错误始终记录在ERROR级别；这是不可修改的。

在容器生命周期的各个阶段还发布了其他一些事件：

• AsyncConsumerStartedEvent(开始使用 Consumer 时)

• AsyncConsumerRestartedEvent(当使用者在失败后重新启动时-仅SimpleMessageListenerContainer)

• AsyncConsumerTerminatedEvent(当 Consumer 正常停止时)

• AsyncConsumerStoppedEvent(当使用者停止时-仅SimpleMessageListenerContainer)

• ConsumeOkEvent(当从代理接收到consumeOk时，包含队列名和consumerTag)

• ListenerContainerIdleEvent(请参阅“检测空闲的异步用户”一节)

Consumer Tags

您可以提供一种生成 Consumer 标签的策略。默认情况下，Consumer 标签将由代理生成。

public interface ConsumerTagStrategy {

    String createConsumerTag(String queue);

}

队列可用，因此可以(可选)在标记中使用该队列。

See 第 3.1.17 节“消息侦听器容器配置”.

注解驱动的侦听器端点

Introduction

异步接收消息的最简单方法是使用带 Comments 的侦听器端点基础结构。简而言之，它允许您将托管 bean 的方法公开为 Rabbit 侦听器端点。

@Component
public class MyService {

    @RabbitListener(queues = "myQueue")
    public void processOrder(String data) {
        ...
    }

}

上面示例的想法是，每当一条消息在名为myQueue的队列上可用时，便会相应地调用processOrder方法(在这种情况下，使用消息的有效负载)。

带 Comments 的端点基础结构使用RabbitListenerContainerFactory在幕后为每种带 Comments 的方法创建一个消息侦听器容器。

在上面的示例中，myQueue必须已经存在并绑定到某些交换。只要应用程序上下文中存在RabbitAdmin，就可以自动声明和绑定队列。

@Component
public class MyService {

  @RabbitListener(bindings = @QueueBinding(
        value = @Queue(value = "myQueue", durable = "true"),
        exchange = @Exchange(value = "auto.exch", ignoreDeclarationExceptions = "true"),
        key = "orderRoutingKey")
  )
  public void processOrder(Order order) {
    ...
  }

  @RabbitListener(bindings = @QueueBinding(
        value = @Queue,
        exchange = @Exchange(value = "auto.exch"),
        key = "invoiceRoutingKey")
  )
  public void processInvoice(Invoice invoice) {
    ...
  }

  @RabbitListener(queuesToDeclare = @Queue(name = "${my.queue}", durable = "true"))
  public String handleWithSimpleDeclare(String data) {
      ...
  }

}

在第一个示例中，如果需要，队列myQueue将与交换机一起自动声明(持久)，并与路由键绑定到交换机。在第二个示例中，将声明并绑定一个匿名(专有，自动删除)队列。可以提供多个QueueBinding条目，从而允许侦听器侦听多个队列。在第三个示例中，如果需要，将声明一个具有从属性my.queue检索的名称的队列，并使用该队列名称作为路由键将默认绑定绑定到默认交换机。

自* 2.0 版以来，@ExchangeComments 支持任何交换类型，包括自定义。请参阅AMQP Concepts文档中的更多信息。

需要更高级的配置时，请使用常规的@Bean定义。

请注意第一个示例中关于 Transaction 所的ignoreDeclarationExceptions。例如，这允许绑定到可能具有不同设置(例如internal)的现有交换机。默认情况下，现有 Transaction 所的属性必须匹配。

从* version 2.0 *开始，您现在可以使用多个路由键将队列绑定到交换：

...
    key = { "red", "yellow" }
...

您还可以在@QueueBindingComments 中为队列，交换和绑定指定参数。例如：

@RabbitListener(bindings = @QueueBinding(
        value = @Queue(value = "auto.headers", autoDelete = "true",
                        arguments = @Argument(name = "x-message-ttl", value = "10000",
                                                type = "java.lang.Integer")),
        exchange = @Exchange(value = "auto.headers", type = ExchangeTypes.HEADERS, autoDelete = "true"),
        arguments = {
                @Argument(name = "x-match", value = "all"),
                @Argument(name = "foo", value = "bar"),
                @Argument(name = "baz")
        })
)
public String handleWithHeadersExchange(String foo) {
    ...
}

请注意，该队列的x-message-ttl参数设置为 10 秒。由于参数类型不是String，因此我们必须指定其类型；在这种情况下Integer。与所有此类声明一样，如果队列已经存在，则参数必须与队列中的参数匹配。对于 Headers 交换，我们设置绑定参数以匹配将 Headersfoo设置为bar且 Headersbaz必须存在任何值的消息。 x-match参数表示必须同时满足两个条件。

参数名称，值和类型可以是属性占位符(${...})或 SpEL 表达式(#{...})。 name必须解析为String； type的表达式必须解析为Class或类的完全限定名称。 value必须解析为可以由DefaultConversionService转换为类型的内容(例如上例中的x-message-ttl)。

如果名称解析为null或空String，则该@Argument将被忽略。

Meta-Annotations

有时您可能想对多个侦听器使用相同的配置。为了减少样板配置，可以使用元 Comments 来创建自己的侦听器 Comments：

@Target({ElementType.TYPE, ElementType.METHOD, ElementType.ANNOTATION_TYPE})
@Retention(RetentionPolicy.RUNTIME)
@RabbitListener(bindings = @QueueBinding(
        value = @Queue,
        exchange = @Exchange(value = "metaFanout", type = ExchangeTypes.FANOUT)))
public @interface MyAnonFanoutListener {
}

public class MetaListener {

    @MyAnonFanoutListener
    public void handle1(String foo) {
        ...
    }

    @MyAnonFanoutListener
    public void handle2(String foo) {
        ...
    }

}

在此示例中，由@MyAnonFanoutListenerComments 创建的每个侦听器都将匿名自动删除队列绑定到扇出交换机metaFanout。元 Comments 机制很简单，因为不检查用户定义 Comments 上的属性-因此您不能覆盖元 Comments 中的设置。当需要更高级的配置时，请使用正常的@Bean定义。

启用侦听器端点 Comments

要启用对@RabbitListenerComments 的支持，请将@EnableRabbit添加到您的@Configuration类中。

@Configuration
@EnableRabbit
public class AppConfig {

    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory() {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory());
        factory.setConcurrentConsumers(3);
        factory.setMaxConcurrentConsumers(10);
        return factory;
    }
}

从* version 2.0 *版本开始，DirectMessageListenerContainerFactory也可用，它将创建DirectMessageListenerContainer s。

默认情况下，基础结构将查找名为rabbitListenerContainerFactory的 bean 作为工厂用来创建消息侦听器容器的源。在这种情况下，无需考虑 RabbitMQ 基础结构设置，就可以使用 3 个线程的核心轮询大小和 10 个线程的最大池大小来调用processOrder方法。

可以自定义侦听器容器工厂以使用每个 Comments，或者可以通过实现RabbitListenerConfigurer接口配置显式默认值。仅当至少一个端点在没有特定容器工厂的情况下注册时才需要使用默认值。有关完整的详细信息和示例，请参见 javadoc。

容器工厂提供添加MessagePostProcessor的方法，这些方法将在接收消息之后(调用侦听器之前)和发送回复之前应用。

从版本 2.0.6 开始，可以将RetryTemplate和RecoveryCallback添加到侦听器容器工厂，用于发送答复。重试用尽时将调用RecoveryCallback；否则，将调用RecoveryCallback。 SendRetryContextAccessor可用于从上下文中获取信息：

factory.setRetryTemplate(retryTemplate);
factory.setReplyRecoveryCallback(ctx -> {
    Message failed = SendRetryContextAccessor.getMessage(ctx);
    Address replyTo = SendRetryContextAccessor.getAddress(ctx);
    Throwable t = ctx.getLastThrowable();
    ...
    return null;
});

如果您喜欢 XML 配置，请使用<rabbit:annotation-driven>元素；将检测到任何带有@RabbitListenerComments 的 bean。

对于SimpleRabbitListenerContainer秒：

<rabbit:annotation-driven/>

<bean id="rabbitListenerContainerFactory"
      class="org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory">
    <property name="connectionFactory" ref="connectionFactory"/>
    <property name="concurrentConsumers" value="3"/>
    <property name="maxConcurrentConsumers" value="10"/>
</bean>

并持续DirectMessageListenerContainer s：

<rabbit:annotation-driven/>

<bean id="rabbitListenerContainerFactory"
      class="org.springframework.amqp.rabbit.config.DirectRabbitListenerContainerFactory">
    <property name="connectionFactory" ref="connectionFactory"/>
    <property name="consumersPerQueue" value="3"/>
</bean>

从* version 2.0 *版本开始，@RabbitListenerComments 具有concurrency属性；它支持 SpEL 表达式(#{...})和属性占位符(${...})。其含义以及允许的值取决于容器类型。

• 对于DirectMessageListenerContainer，该值必须是单个整数值，该值设置了容器上的consumersPerQueue属性。

• 对于SimpleRabbitListenerContainer，该值可以是单个整数值，该值在容器上设置concurrentConsumers属性，也可以采用m-n的形式，其中m是concurrentConsumers属性，而n是maxConcurrentConsumers属性。

无论哪种情况，此设置都将覆盖出厂设置。以前，如果您的侦听器需要不同的并发性，则必须定义不同的容器工厂。

Comments 方法的消息转换

调用侦听器之前，管道中有两个转换步骤。第一个使用MessageConverter将传入的 Spring AMQP Message转换为* spring-messaging * Message。调用目标方法时，如有必要，消息有效负载将转换为方法参数类型。

第一步的默认MessageConverter是 Spring AMQP SimpleMessageConverter，它处理到String和java.io.Serializable对象的转换。所有其他都保留为byte[]。在下面的讨论中，我们将此称为消息转换器。

第二步的默认转换器是GenericMessageConverter，它委派给转换服务(DefaultFormattingConversionService的实例)。在下面的讨论中，我们将此称为方法参数转换器。

要更改* message converter *，只需将其作为属性添加到容器工厂 bean 中：

@Bean
public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory() {
    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
    ...
    factory.setMessageConverter(new Jackson2JsonMessageConverter());
    ...
    return factory;
}

这将配置一个 Jackson2 转换器，该转换器希望出现 Headers 信息以指导转换。

您也可以考虑使用ContentTypeDelegatingMessageConverter，它可以处理不同 Content Type 的转换。

在大多数情况下，除非必须使用自定义ConversionService，否则不必自定义* method arguments converter *。

在* 1.6 之前的版本中，必须在消息头中提供用于转换 JSON 的类型信息，或者需要自定义ClassMapper。从 version 1.6 *开始，如果没有类型信息 Headers，则可以从目标方法参数中推断类型。

有关更多信息，请参见名为“ Jackson2JsonMessageConverter”的部分。

如果您希望自定义* method arguments converter *，则可以按照以下步骤进行：

@Configuration
@EnableRabbit
public class AppConfig implements RabbitListenerConfigurer {

    ...

    @Bean
    public DefaultMessageHandlerMethodFactory myHandlerMethodFactory() {
        DefaultMessageHandlerMethodFactory factory = new DefaultMessageHandlerMethodFactory();
        factory.setMessageConverter(new GenericMessageConverter(myConversionService()));
        return factory;
    }

    @Bean
    public ConversionService myConversionService() {
        DefaultConversionService conv = new DefaultConversionService();
        conv.addConverter(mySpecialConverter());
        return conv;
    }

    @Override
    public void configureRabbitListeners(RabbitListenerEndpointRegistrar registrar) {
        registrar.setMessageHandlerMethodFactory(myHandlerMethodFactory());
    }

    ...

}

程序化端点注册

RabbitListenerEndpoint提供 Rabbit 端点的模型，并负责为该模型配置容器。除了RabbitListenerComments 检测到的端点外，基础结构还允许您以编程方式配置端点。

@Configuration
@EnableRabbit
public class AppConfig implements RabbitListenerConfigurer {

    @Override
    public void configureRabbitListeners(RabbitListenerEndpointRegistrar registrar) {
        SimpleRabbitListenerEndpoint endpoint = new SimpleRabbitListenerEndpoint();
        endpoint.setQueueNames("anotherQueue");
        endpoint.setMessageListener(message -> {
            // processing
        });
        registrar.registerEndpoint(endpoint);
    }
}

在上面的示例中，我们使用了SimpleRabbitListenerEndpoint来提供实际的MessageListener进行调用，但是您也可以构建自己的描述自定义调用机制的端点变量。

应该注意的是，您也可以完全跳过@RabbitListener的使用，而仅通过 RabbitListenerConfigurer 以编程方式注册端点。

带 Comments 的端点方法签名

到目前为止，我们已经在端点中注入了一个简单的 String，但实际上它可以具有非常灵活的方法签名。让我们重写它，以使用自定义 Headers 注入Order：

@Component
public class MyService {

    @RabbitListener(queues = "myQueue")
    public void processOrder(Order order, @Header("order_type") String orderType) {
        ...
    }
}

这些是您可以在侦听器端点中注入的主要元素：

原始的org.springframework.amqp.core.Message。

收到消息的com.rabbitmq.client.Channel

org.springframework.messaging.Message代表传入的 AMQP 消息。请注意，此消息同时包含自定义 Headers 和标准 Headers(由AmqpHeaders定义)。

@Header-带 Comments 的方法参数以提取特定的 Headers 值，包括标准 AMQPHeaders。

@Headers-带 Comments 的参数，也必须可分配给java.util.Map以访问所有 Headers。

不是受支持类型之一(即Message和Channel)的未 Comments 元素被视为有效负载。您可以通过用@PayloadComments 参数来使其明确。您还可以通过添加额外的@Valid来启用验证。

注入 Spring 的 Message 抽象的能力特别有用，它可以受益于存储在特定于传输的消息中的所有信息，而无需依赖于特定于传输的 API。

@RabbitListener(queues = "myQueue")
public void processOrder(Message<Order> order) { ...
}

DefaultMessageHandlerMethodFactory提供了方法参数的处理，可以进一步对其进行自定义以支持其他方法参数。转换和验证支持也可以在那里定制。

例如，如果要在处理订单之前确保其有效，则可以使用@ValidComments 有效负载，并按以下方式配置必要的验证器：

@Configuration
@EnableRabbit
public class AppConfig implements RabbitListenerConfigurer {

    @Override
    public void configureRabbitListeners(RabbitListenerEndpointRegistrar registrar) {
        registrar.setMessageHandlerMethodFactory(myHandlerMethodFactory());
    }

    @Bean
    public DefaultMessageHandlerMethodFactory myHandlerMethodFactory() {
        DefaultMessageHandlerMethodFactory factory = new DefaultMessageHandlerMethodFactory();
        factory.setValidator(myValidator());
        return factory;
    }
}

收听多个队列

使用queues属性时，可以指定关联的容器可以侦听多个队列。您可以使用@HeaderComments 使从中接收消息的队列名称可用于 POJO 方法：

@Component
public class MyService {

    @RabbitListener(queues = { "queue1", "queue2" } )
    public void processOrder(String data, @Header(AmqpHeaders.CONSUMER_QUEUE) String queue) {
        ...
    }

}

从* version 1.5 *版本开始，您可以使用属性占位符和 SpEL 来外部化队列名称：

@Component
public class MyService {

    @RabbitListener(queues = "#{'${property.with.comma.delimited.queue.names}'.split(',')}" )
    public void processOrder(String data, @Header(AmqpHeaders.CONSUMER_QUEUE) String queue) {
        ...
    }

}

在 1.5 版之前，只能以这种方式指定一个队列；每个队列都需要一个单独的属性。

Reply Management

MessageListenerAdapter中的现有支持已经允许您的方法具有非空返回类型。在这种情况下，调用的结果将封装在一条消息中，该消息要么以原始消息的ReplyToAddressHeaders 中指定的地址发送，要么以侦听器上配置的默认地址发送。现在可以使用消息传递抽象的@SendToComments 设置默认地址。

假设我们的processOrder方法现在应该返回OrderStatus，则可以按照以下方式编写它以自动发送回复：

@RabbitListener(destination = "myQueue")
@SendTo("status")
public OrderStatus processOrder(Order order) {
    // order processing
    return status;
}

如果需要以与传输无关的方式设置其他 Headers，则可以返回Message，例如：

@RabbitListener(destination = "myQueue")
@SendTo("status")
public Message<OrderStatus> processOrder(Order order) {
    // order processing
    return MessageBuilder
        .withPayload(status)
        .setHeader("code", 1234)
        .build();
}

@SendTo值假定为模式exchange/routingKey之后的回复exchange和routingKey对，其中那些部分之一可以省略。有效值为：

foo/bar-回复要交换和路由的密钥。

foo/-replyTo 交换和默认(空)routingKey。

bar或/bar-ReplyTo routingKey 和默认(空)交换。

/或为空-ReplyTo 默认交换和默认 routingKey。

@SendTo也可以不带value属性使用。这种情况下等于一个空的 sendTo 模式。仅当入站邮件没有replyToAddress属性时才使用@SendTo。

从* version 1.5 *开始，@SendTo值可以是 bean 初始化 SpEL 表达式，例如…

@RabbitListener(queues = "test.sendTo.spel")
@SendTo("#{spelReplyTo}")
public String capitalizeWithSendToSpel(String foo) {
    return foo.toUpperCase();
}
...
@Bean
public String spelReplyTo() {
    return "test.sendTo.reply.spel";
}

该表达式的计算结果必须为String，它可以是简单的队列名称(发送到默认交换机)或采用如上所述的exchange/routingKey形式。

对于动态回复路由，消息发送者应包括reply_to消息属性或使用下面描述的备用运行时 SpEL 表达式。

从* version 1.6 *开始，@SendTo可以是 SpEL 表达式，它会在运行时针对请求和回复进行评估：

@RabbitListener(queues = "test.sendTo.spel")
@SendTo("!{'some.reply.queue.with.' + result.queueName}")
public Bar capitalizeWithSendToSpel(Foo foo) {
    return processTheFooAndReturnABar(foo);
}

SpEL 表达式的运行时性质由!{...}分隔符指示。表达式的评估上下文#root对象具有三个属性：

• request-o.s.amqp.core.Message请求对象。

• source-转换后的o.s.messaging.Message<?>。

• result-方法结果。

该上下文具有一个 map 属性访问器，一个标准类型转换器和一个 bean 解析器，从而允许引用上下文中的其他 bean(例如@someBeanName.determineReplyQ(request, result))。

总而言之，#{...}在初始化期间被评估一次，其中#root对象是应用程序上下文； bean 由它们的名称引用。对于每条消息，在运行时都会对!{...}进行评估，其根对象具有上述属性，并且使用其名称(以@作为前缀)引用 bean。

从 2.1 版开始，${some.reply.to}也支持简单的属性占位符。对于早期版本，可以将以下内容用作解决方法：

@RabbitListener(queues = "foo")
@SendTo("#{environment['my.send.to']}")
public String listen(Message in) {
    ...
    return ...
}

Multi-Method Listeners

从* version 1.5.0 *开始，现在可以在类级别指定@RabbitListenerComments。与新的@RabbitHandler注解一起，这允许单个侦听器根据传入消息的有效负载类型来调用不同的方法。最好用一个例子来描述：

@RabbitListener(id="multi", queues = "someQueue")
@SendTo("my.reply.queue")
public class MultiListenerBean {

    @RabbitHandler
    public String bar(Bar bar) {
        ...
    }

    @RabbitHandler
    public String baz(Baz baz) {
        ...
    }

    @RabbitHandler
    public String qux(@Header("amqp_receivedRoutingKey") String rk, @Payload Qux qux) {
        ...
    }

    @RabbitHandler(isDefault = true)
    public String defaultMethod(Object object) {
        ...
    }

}

在这种情况下，如果转换后的有效负载是Bar，Baz或Qux，则将调用单独的@RabbitHandler方法。重要的是要了解系统必须能够基于有效负载类型识别唯一方法。检查该类型是否可分配给没有 Comments 或带有@PayloadComments 的单个参数。注意，与上述方法级别@RabbitListener中讨论的方法签名相同。

从* version 2.0.3 *开始，可以将@RabbitHandler方法指定为默认方法，如果其他方法不匹配，则会调用该方法。最多可以指定一种方法。

@Repeatable @RabbitListener

从* version 1.6 *开始，@RabbitListenerComments 用@Repeatable标记。这意味着 Comments 可以多次出现在相同的 Comments 元素(方法或类)上。在这种情况下，将为每个 Comments 创建一个单独的侦听器容器，每个 Comments 都调用相同的侦听器@Bean。 Java 8 或更高版本可以使用可重复的注解；当使用 Java 7 或更早版本时，通过使用@RabbitListeners“容器”Comments 以及@RabbitListenerComments 数组可以达到相同的效果。

代理@RabbitListener 和泛型

如果打算代理您的服务(例如@Transactional)，则在接口具有通用参数时需要注意一些事项。具有通用接口和特定实现，例如：

interface TxService<P> {

   String handle(P payload, String header);

}

static class TxServiceImpl implements TxService<Foo> {

    @Override
    @RabbitListener(...)
    public String handle(Foo foo, String rk) {
         ...
    }

}

您必须切换到 CGLIB 目标类代理，因为接口handle方法的实际实现是 bridge 方法。对于事务 Management，使用 Comments 选项@EnableTransactionManagement(proxyTargetClass = true)来配置 CGLIB 的使用。在这种情况下，必须在实现中的目标方法上声明所有 Comments：

static class TxServiceImpl implements TxService<Foo> {

    @Override
    @Transactional
    @RabbitListener(...)
    public String handle(@Payload Foo foo, @Header("amqp_receivedRoutingKey") String rk) {
        ...
    }

}

Handling Exceptions

默认情况下，如果带 Comments 的侦听器方法引发异常，则将其引发到容器中，并将根据容器和代理的配置将消息重新排队并重新传递，丢弃或路由到 Dead Letter Exchange。什么也没有退还给发件人。

从* version 2.0 *版本开始，@RabbitListenerComments 具有两个新属性：errorHandler和returnExceptions。

这些默认情况下未配置。

使用errorHandler提供RabbitListenerErrorHandler实现的 bean 名称。此功能接口有一种方法：

@FunctionalInterface
public interface RabbitListenerErrorHandler {

    Object handleError(Message amqpMessage, org.springframework.messaging.Message<?> message,
              ListenerExecutionFailedException exception) throws Exception;

}

如您所见，您可以访问从容器接收的原始消息，消息转换器生成的 spring-messaging Message<?>对象以及包装在ListenerExecutionFailedException中的由侦听器引发的异常。根据returnExceptions的设置，错误处理程序可以返回将作为答复发送的某些结果，也可以引发将原始或新的异常抛出给容器或返回给发送者。

returnExceptions属性为“ true”时，将导致异常返回给发送者。异常包装在RemoteInvocationResult对象中。在发送方，有一个可用的RemoteInvocationAwareMessageConverterAdapter，如果配置为RabbitTemplate，它将重新引发服务器端异常，并包装在AmqpRemoteException中。服务器异常的堆栈跟踪将通过合并服务器和 Client 端堆栈跟踪进行综合。

Container Management

为 Comments 创建的容器未在应用程序上下文中注册。您可以通过在RabbitListenerEndpointRegistry bean 上调用getListenerContainers()获得所有容器的集合。然后，您可以遍历此集合，例如，停止/启动所有容器或在注册表本身上调用Lifecycle方法，这将在每个容器上调用操作。

您还可以使用id，getListenerContainer(String id)获得对单个容器的引用；例如registry.getListenerContainer("multi")(由上面的代码段创建的容器)。

从* 1.5.2 *版本开始，您可以使用getListenerContainerIds()获得id s 已注册容器。

从* version 1.5 *开始，您现在可以将group分配给RabbitListener端点上的容器。这提供了一种获取容器子集引用的机制。添加group属性会使类型为Collection<MessageListenerContainer>的 bean 注册到具有组名的上下文中。

使用集装箱工厂

引入了侦听器容器工厂以支持@RabbitListener和/或向RabbitListenerEndpointRegistry注册容器，如“程序化端点注册”部分中所述。

从* 2.1 *版本开始，它们可用于创建任何侦听器容器。甚至没有侦听器的容器(例如用于 Spring Integration 中的容器)。当然，必须在启动容器之前添加侦听器。

有两种创建此类容器的方法：

使用 SimpleRabbitListenerEndpoint.

@Bean
public SimpleMessageListenerContainer factoryCreatedContainerSimpleListener(
        SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory) {
    SimpleRabbitListenerEndpoint endpoint = new SimpleRabbitListenerEndpoint();
    endpoint.setQueueNames("queue.1");
    endpoint.setMessageListener(message -> {
        ...
    });
    return rabbitListenerContainerFactory.createListenerContainer(endpoint);
}

创建后添加侦听器.

@Bean
public SimpleMessageListenerContainer factoryCreatedContainerNoListener(
        SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory) {
    SimpleMessageListenerContainer container = rabbitListenerContainerFactory.createListenerContainer();
    container.setMessageListener(message -> {
        ...
    });
    container.setQueueNames("test.no.listener.yet");
    return container;
}

无论哪种情况，监听器都可以是ChannelAwareMessageListener，因为它现在是MessageListener的子接口。

如果您希望创建多个具有相似属性的容器和/或使用预先配置的容器工厂(例如 Spring Boot 自动配置提供的容器工厂)，则这些技术很有用。

异步@RabbitListener 返回类型

从版本 2.1 开始，可以使用异步返回类型ListenableFuture<?>和Mono<?>指定@RabbitListener(和@RabbitHandler)方法，从而允许异步发送回复。

线程和异步使用者

异步使用者涉及许多不同的线程。

当RabbitMQ Client传递新消息时，将使用SimpleMessageListenerContainer中配置的TaskExecutor的线程来调用MessageListener。如果未配置，则使用SimpleAsyncTaskExecutor。如果使用池 Actuator，请确保池大小足以处理配置的并发。使用DirectMessageListenerContainer，可以直接在RabbitMQ Client线程上调用MessageListener。在这种情况下，taskExecutor用于监视使用者的任务。

创建连接时，在CachingConnectionFactory中配置的Executor被传递到RabbitMQ Client，并且其线程用于将新消息传递到侦听器容器。在编写本文时，如果未配置它，则 Client 端使用内部线程池执行程序，该执行程序的池大小为 5.

RabbitMQ client使用ThreadFactory为低级 I/O(套接字)操作创建线程。要修改此工厂，您需要配置基础 RabbitMQ ConnectionFactory，如名为“配置基础 Client 端连接工厂”的部分中所述。

选择一个容器

• 2.0 版*引入了DirectMessageListenerContainer(DMLC)；以前，只有SimpleMessageListenerContainer(SMLC)可用。 SMLC 为每个使用者使用内部队列和专用线程。如果将一个容器配置为侦听多个队列，则使用相同的使用者线程来处理所有队列。并发由concurrentConsumers和其他属性控制。当消息从 RabbitMQClient 端到达时，Client 端线程将它们通过队列传递给使用方线程。之所以需要这种架构，是因为在 RabbitMQClient 端的早期版本中，不可能同时进行多个交付。较新版本的 Client 端具有修订的线程模型，现在可以支持并发。这允许引入 DMLC，现在可以在 RabbitMQClient 端线程上直接调用侦听器。因此，它的体系结构实际上比 SMLC“简单”。但是，这种方法存在一些局限性，DMLC 无法使用 SMLC 的某些功能。并发也由consumersPerQueue(和 Client 端库的线程池)控制； concurrentConsumers和关联的属性不适用于此容器。

SMLC 可用以下功能，而 DMLC 不可用：

• txSize-使用 SMLC，您可以将其设置为控制事务中传递的消息数量和/或减少确认的数量，但是在失败后可能导致重复传递的数量增加。 (DMLC 确实具有messagesPerAck，可用于减少错误，与txSize和 SMLC 相同，但不能与事务一起使用-每条消息均在单独的事务中传递和确认)。

• maxConcurrentConsumers和使用者缩放间隔/触发-DMLC 中没有自动缩放；但是，它确实允许您以编程方式更改consumersPerQueue属性，并且将相应地调整使用者。

但是，DMLC 比 SMLC 具有以下优点：

• 在运行时添加和删除队列更有效；通过 SMLC，整个使用者线程将重新启动(所有使用者都被取消并重新创建)；使用 DMLC，不会取消未受影响的 Consumer。

• 避免在 RabbitMQClient 端线程和使用者线程之间进行上下文切换。

• 线程在使用者之间共享，而不是在 SMLC 中为每个使用者使用专用线程。但是，请参阅称为“线程和异步使用者”的部分中有关连接工厂配置的重要说明。

有关将哪些配置属性应用于每个容器的信息，请参见第 3.1.17 节“消息侦听器容器配置”。

检测空闲的异步使用者

尽管效率很高，但异步使用者的一个问题是检测它们何时处于空闲状态-如果一段时间内没有消息到达，则用户可能希望采取某些措施。

从* version 1.6 *开始，现在可以将侦听器容器配置为在一段时间没有消息传递的情况下发布ListenerContainerIdleEvent。容器空闲时，每idleEventInterval毫秒将发布一次事件。

要配置此功能，请在容器上设置idleEventInterval：

xml

<rabbit:listener-container connection-factory="connectionFactory"
        ...
        idle-event-interval="60000"
        ...
        >
    <rabbit:listener id="container1" queue-names="foo" ref="myListener" method="handle" />
</rabbit:listener-container>

Java

@Bean
public SimpleMessageListenerContainer(ConnectionFactory connectionFactory) {
    SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
    ...
    container.setIdleEventInterval(60000L);
    ...
    return container;
}

@RabbitListener

@Bean
public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory() {
    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
    factory.setConnectionFactory(rabbitConnectionFactory());
    factory.setIdleEventInterval(60000L);
    ...
    return factory;
}

在每种情况下，容器闲置时每分钟将发布一次事件。

Event Consumption

您可以通过实现ApplicationListener捕获这些事件-可以是常规侦听器，也可以是仅侦听仅接收此特定事件的监听器。您还可以使用 Spring Framework 4.2 中引入的@EventListener。

以下示例将@RabbitListener和@EventListener合并为一个类。重要的是要了解，应用程序侦听器将获取所有容器的事件，因此，如果您要基于哪个容器处于空闲状态采取特定的操作，则可能需要检查侦听器 ID。您也可以为此使用@EventListener condition。

事件具有 4 个属性：

• source-侦听器容器实例

• id-侦听器 ID(或容器 Bean 名称)

• idleTime-事件发布时容器空闲的时间

• queueNames-容器侦听的队列的名称

public class Listener {

    @RabbitListener(id="foo", queues="#{queue.name}")
    public String listen(String foo) {
        return foo.toUpperCase();
    }

    @EventListener(condition = "event.listenerId == 'foo'")
    public void onApplicationEvent(ListenerContainerIdleEvent event) {
        ...
    }

}

3.1.7 容器和以代理命名的队列

虽然最好使用AnonymousQueues作为自动删除队列，但从* version 2.1 *开始，您可以将具有代理名称的队列与侦听器容器一起使用：

@Bean
public Queue queue() {
    return new Queue("", false, true, true);
}

@Bean
public SimpleMessageListenerContainer container() {
    SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(cf());
    container.setQueues(queue());
    container.setMessageListener(m -> {
        ...
    });
    container.setMissingQueuesFatal(false);
    return container;
}

请注意，该名称为空字符串。 RabbitAdmin声明队列时，它将使用代理返回的名称更新Queue.actualName属性。在配置容器时，必须使用setQueues()，此功能才能起作用，以便容器可以在运行时访问声明的名称，仅设置名称是不够的。

3.1.8 消息转换器

Introduction

AmqpTemplate还定义了几种发送和接收将委派给MessageConverter的消息的方法。 MessageConverter本身非常简单。它为每个方向提供了一种方法：一种用于将消息转换成消息，而另一种用于将消息转换成消息。请注意，转换为消息时，除了对象之外，您还可以提供属性。 “对象”参数通常对应于消息正文。

public interface MessageConverter {

    Message toMessage(Object object, MessageProperties messageProperties)
            throws MessageConversionException;

    Object fromMessage(Message message) throws MessageConversionException;

}

下面列出了AmqpTemplate上的相关消息发送方法。它们比我们之前讨论的方法更简单，因为它们不需要Message实例。相反，MessageConverter负责通过将提供的对象转换为Message主体的字节数组，然后添加任何提供的MessageProperties来“创建”每个Message。

void convertAndSend(Object message) throws AmqpException;

void convertAndSend(String routingKey, Object message) throws AmqpException;

void convertAndSend(String exchange, String routingKey, Object message)
    throws AmqpException;

void convertAndSend(Object message, MessagePostProcessor messagePostProcessor)
    throws AmqpException;

void convertAndSend(String routingKey, Object message,
    MessagePostProcessor messagePostProcessor) throws AmqpException;

void convertAndSend(String exchange, String routingKey, Object message,
    MessagePostProcessor messagePostProcessor) throws AmqpException;

在接收方，只有两种方法：一种接受队列名称，另一种依赖于已设置的模板的“ queue”属性。

Object receiveAndConvert() throws AmqpException;

Object receiveAndConvert(String queueName) throws AmqpException;

SimpleMessageConverter

MessageConverter策略的默认实现称为SimpleMessageConverter。如果您未明确配置替代方案，那么 RabbitTemplate 实例将使用此转换器。它处理基于文本的内容，序列化的 Java 对象和简单的字节数组。

从邮件转换

如果 ImportingMessage 的 Content Type 以“ text”开头(例如“ text/plain”)，则还将检查 content-encoding 属性，以确定将 Message 主体字节数组转换为 Java String 时要使用的字符集。 。如果未在 Importing 消息上设置任何内容编码属性，则默认情况下它将使用“ UTF-8”字符集。如果需要覆盖该默认设置，则可以配置SimpleMessageConverter的实例，设置其“ defaultCharset”属性，然后将其注入RabbitTemplate实例。

如果 ImportingMessage 的 content-type 属性值设置为“ application/x-java-serialized-object”，则SimpleMessageConverter将尝试将字节数组反序列化(重新水化)为 Java 对象。虽然这可能对简单的原型制作很有用，但通常不建议您依赖 Java 序列化，因为它会导致生产者和使用者之间的紧密耦合。当然，它也排除了使用非 Java 系统的任何一方。由于 AMQP 是线级协议，因此不幸的是，由于此类限制而失去了很多优势。在接下来的两节中，我们将探讨一些无需依赖 Java 序列化即可传递富域对象内容的替代方法。

对于所有其他 Content Type，SimpleMessageConverter将直接以字节数组形式返回消息正文内容。

有关重要信息，请参见称为“ Java 反序列化”的部分。

转换为邮件

从任意 Java 对象转换为消息时，SimpleMessageConverter同样处理字节数组，字符串和可序列化的实例。它将把这些转换成字节(对于字节数组，没有任何转换)，并将相应地设置 content-type 属性。如果要转换的对象与这些类型之一不匹配，则消息正文将为 null。

SerializerMessageConverter

该转换器与SimpleMessageConverter相似，不同之处在于它可以与其他 Spring Framework Serializer和Deserializer实现一起进行application/x-java-serialized-object转换配置。

有关重要信息，请参见称为“ Java 反序列化”的部分。

Jackson2JsonMessageConverter

转换为邮件

如前一节所述，通常不建议依赖 Java 序列化。 JSON(JavaScript 对象表示法)是一种更常见且更灵活且可跨不同语言和平台移植的替代方法。可以在任何RabbitTemplate实例上配置该转换器，以覆盖其对SimpleMessageConverter默认实例的使用。 Jackson2JsonMessageConverter使用com.fasterxml.jackson 2.x 库。

<bean class="org.springframework.amqp.rabbit.core.RabbitTemplate">
    <property name="connectionFactory" ref="rabbitConnectionFactory"/>
    <property name="messageConverter">
        <bean class="org.springframework.amqp.support.converter.Jackson2JsonMessageConverter">
            <!-- if necessary, override the DefaultClassMapper -->
            <property name="classMapper" ref="customClassMapper"/>
        </bean>
    </property>
</bean>

如上所示，默认情况下Jackson2JsonMessageConverter使用DefaultClassMapper。类型信息被添加到MessageProperties(并从中检索)。如果入站邮件的MessageProperties中不包含类型信息，但是您知道所需的类型，则可以使用defaultType属性配置静态类型。

<bean id="jsonConverterWithDefaultType"
      class="o.s.amqp.support.converter.Jackson2JsonMessageConverter">
    <property name="classMapper">
        <bean class="org.springframework.amqp.support.converter.DefaultClassMapper">
            <property name="defaultType" value="foo.PurchaseOrder"/>
        </bean>
    </property>
</bean>

此外，您可以根据__TypeId__Headers 中的值提供自定义 Map…

@Bean
public Jackson2JsonMessageConverter jsonMessageConverter() {
    Jackson2JsonMessageConverter jsonConverter = new Jackson2JsonMessageConverter();
    jsonConverter.setClassMapper(classMapper());
    return jsonConverter;
}

@Bean
public DefaultClassMapper classMapper() {
    DefaultClassMapper classMapper = new DefaultClassMapper();
    Map<String, Class<?>> idClassMapping = new HashMap<>();
    idClassMapping.put("foo", Foo.class);
    idClassMapping.put("bar", Bar.class);
    classMapper.setIdClassMapping(idClassMapping);
    return classMapper;
}

现在，如果发送系统将 Headers 设置为foo，则转换器将创建Foo对象，依此类推。有关转换来自非 Spring 应用程序的消息的完整讨论，请参见第 3.3.4 节“从非 Spring 应用程序接收 JSON”示例应用程序。

从邮件转换

根据发送系统添加到 Headers 的类型信息，将入站消息转换为对象。

在* 1.6 之前的版本中，如果不存在类型信息，则转换将失败。从 version 1.6 *开始，如果缺少类型信息，则转换器将使用 Jackson 的默认值(通常是 Map)转换 JSON。

同样，从* version 1.6 *开始，当使用@RabbitListenerComments(在方法上)时，推断的类型信息将添加到MessageProperties；这使转换器可以转换为目标方法的参数类型。仅当存在一个不带 Comments 的参数或一个带有@PayloadComments 的参数时，才适用。在分析期间会忽略类型为Message的参数。

@RabbitListener
public void foo(Foo foo) {...}

@RabbitListener
public void foo(@Payload Foo foo, @Header("amqp_consumerQueue") String queue) {...}

@RabbitListener
public void foo(Foo foo, o.s.amqp.core.Message message) {...}

@RabbitListener
public void foo(Foo foo, o.s.messaging.Message<Foo> message) {...}

@RabbitListener
public void foo(Foo foo, String bar) {...}

@RabbitListener
public void foo(Foo foo, o.s.messaging.Message<?> message) {...}

在上面的前四种情况下，转换器将尝试转换为Foo类型。第五个示例无效，因为我们无法确定哪个参数应接收消息有效负载。对于第六个示例，由于通用类型为WildcardType，因此将应用 Jackson 的默认值。

但是，您可以创建一个自定义转换器，并使用targetMethod message 属性来确定将 JSON 转换为哪种类型。

使用 RabbitTemplate 从消息转换

如上所述，类型信息在消息头中传送，以在从消息转换时帮助转换器。这在大多数情况下都可以正常工作，但是在使用泛型类型时，它只能转换简单对象和已知的“容器”对象(列表，数组，Map)。从 2.0 版开始，Jackson2JsonMessageConverter实现SmartMessageConverter，使其可以与带有ParameterizedTypeReference参数的新RabbitTemplate方法一起使用；这允许转换复杂的泛型类型。例如：

Foo<Bar<Baz, Qux>> foo =
    rabbitTemplate.receiveAndConvert(new ParameterizedTypeReference<Foo<Bar<Baz, Qux>>>() { });

MarshallingMessageConverter

另一个选择是MarshallingMessageConverter。它委派给Marshaller和Unmarshaller策略接口的 Spring OXM 库实现。您可以阅读有关该库here的更多信息。就配置而言，仅提供构造函数参数是最常见的，因为Marshaller的大多数实现也将实现Unmarshaller。

<bean class="org.springframework.amqp.rabbit.core.RabbitTemplate">
    <property name="connectionFactory" ref="rabbitConnectionFactory"/>
    <property name="messageConverter">
        <bean class="org.springframework.amqp.support.converter.MarshallingMessageConverter">
            <constructor-arg ref="someImplemenationOfMarshallerAndUnmarshaller"/>
        </bean>
    </property>
</bean>

Jackson2XmlMessageConverter

此类是在* 2.1 版中引入的，可用于在 XML 之间转换消息。

Jackson2XmlMessageConverter和Jackson2JsonMessageConverter都具有相同的 Base ClassAbstractJackson2MessageConverter。

Jackson2XmlMessageConverter使用com.fasterxml.jackson 2.x 库。

您可以使用与Jackson2JsonMessageConverter相同的方式来使用它，除了它支持 XML 而不是 JSON。

<bean id="xmlConverterWithDefaultType"
        class="org.springframework.amqp.support.converter.Jackson2XmlMessageConverter">
    <property name="classMapper">
        <bean class="org.springframework.amqp.support.converter.DefaultClassMapper">
            <property name="defaultType" value="foo.PurchaseOrder"/>
        </bean>
    </property>
</bean>

有关更多信息，请参见名为“ Jackson2JsonMessageConverter”的部分。

ContentTypeDelegatingMessageConverter

此类是在*版本 1.4.2 *中引入的，并允许基于MessageProperties中的 content type 属性委派到特定的MessageConverter。默认情况下，如果没有contentType属性，或者没有与任何已配置的转换器匹配的值，它将委派给SimpleMessageConverter。

<bean id="contentTypeConverter" class="ContentTypeDelegatingMessageConverter">
    <property name="delegates">
        <map>
            <entry key="application/json" value-ref="jsonMessageConverter" />
            <entry key="application/xml" value-ref="xmlMessageConverter" />
        </map>
    </property>
</bean>

Java Deserialization

邮件属性转换器

MessagePropertiesConverter策略接口用于在 Rabbit Client BasicProperties和 Spring AMQP MessageProperties之间进行转换。默认实现(DefaultMessagePropertiesConverter)通常可以满足大多数目的，但是您可以根据需要实现自己的实现。当大小不超过1024字节时，默认属性转换器会将LongString类型的BasicProperties元素转换为String。较大的LongString不进行转换(请参见下文)。可以使用构造函数参数来覆盖此限制。

从* version 1.6 *开始，长于长字符串限制(默认值为 1024)的 Headers 现在默认由DefaultMessagePropertiesConverter保留为LongString。您可以通过getBytes[]，toString()或getStream()方法访问内容。

以前，DefaultMessagePropertiesConverter将此类 Headers“转换”为DataInputStream(实际上只是引用了LongString的DataInputStream)。在输出时，此 Headers 未转换(除非转换为字符串，例如通过在流上调用toString()来转换为[email protected])。

现在，大的传入LongStringHeaders 也可以在输出中正确“转换”(默认情况下)。

提供了一个新的构造函数，使您可以配置转换器以像以前一样工作：

/**
 * Construct an instance where LongStrings will be returned
 * unconverted or as a java.io.DataInputStream when longer than this limit.
 * Use this constructor with 'true' to restore pre-1.6 behavior.
 * @param longStringLimit the limit.
 * @param convertLongLongStrings LongString when false,
 * DataInputStream when true.
 * @since 1.6
 */
public DefaultMessagePropertiesConverter(int longStringLimit, boolean convertLongLongStrings) { ... }

同样从* version 1.6 *开始，新属性correlationIdString已添加到MessageProperties。以前，当从 RabbitMQClient 端使用BasicProperties来回转换时，执行了不必要的byte[] <-> String转换，因为MessageProperties.correlationId是byte[]但BasicProperties使用String。 (最终，RabbitMQClient 端使用 UTF-8 将 String 转换为字节以放入协议消息中)。

为了提供最大的向后兼容性，已将新属性correlationIdPolicy添加到DefaultMessagePropertiesConverter。这需要一个DefaultMessagePropertiesConverter.CorrelationIdPolicy枚举参数。默认情况下，它设置为BYTES，它复制了以前的行为。

对于入站邮件：

• STRING-仅 Map 了correlationIdString属性

• BYTES-仅 Map 了correlationId属性

• BOTH-Map 了两个属性

对于出站邮件：

• STRING-仅 Map 了correlationIdString属性

• BYTES-仅 Map 了correlationId属性

• BOTH-将同时考虑这两个属性，其中 String 属性优先

同样从* version 1.6 *开始，入站deliveryMode属性不再 Map 到MessageProperties.deliveryMode，而是 Map 到MessageProperties.receivedDeliveryMode。同样，入站userId属性不再 Map 到MessageProperties.userId，而是 Map 到MessageProperties.receivedUserId。如果将相同的MessageProperties对象用于出站消息，则这些更改是为了避免这些属性的意外传播。

3.1.9 修改邮件-压缩等

存在许多扩展点，您可以在其中对消息执行某些处理，可以将其发送到 RabbitMQ 之前，也可以在接收到消息之后立即进行处理。

从第 3.1.8 节“消息转换器”可以看出，这样的扩展点位于AmqpTemplate convertAndReceive操作中，您可以在其中提供MessagePostProcessor。例如，在转换 POJO 之后，使用MessagePostProcessor可以在Message上设置自定义 Headers 或属性。

从*版本 1.4.2 *开始，附加的扩展点已添加到RabbitTemplate-setBeforePublishPostProcessors()和setAfterReceivePostProcessors()。第一个使后处理器能够在发送到 RabbitMQ 之前立即运行。使用批处理(请参见称为“批处理”的部分)时，将在组装批处理之后和发送批处理之前调用此方法。收到消息后立即调用第二个。

这些扩展点用于压缩等功能，为此，提供了多个MessagePostProcessor：

• GZipPostProcessor

• ZipPostProcessor

用于在发送之前压缩消息，以及

• GUnzipPostProcessor

• UnzipPostProcessor

用于解压缩收到的消息。

同样，SimpleMessageListenerContainer也具有setAfterReceivePostProcessors()方法，从而允许在容器接收到消息后执行解压缩。

3.1.10 请求/回复消息

Introduction

AmqpTemplate还提供了各种sendAndReceive方法，这些方法接受与单向发送操作(交换，routingKey 和 Message)相同的自变量。这些方法对于请求/答复方案非常有用，因为它们在发送之前处理必要的“答复”属性的配置，并且可以在为此目的内部创建的互斥队列上侦听答复消息。

将MessageConverter应用于请求和答复时，也可以使用类似的请求/答复方法。这些方法被命名为convertSendAndReceive。有关更多详细信息，请参见AmqpTemplate的 Javadoc。

从版本 1.5.0 开始，每个sendAndReceive方法变量都有一个重载版本CorrelationData。与正确配置的连接工厂一起，这可以使发布者收到操作的发送方的确认。有关更多信息，请参见称为“发布者确认并返回”的部分和RabbitOperations的 javadoc。

从 2.0 版开始，这些方法(convertSendAndReceiveAsType)的变体需要一个附加的ParameterizedTypeReference参数来转换复杂的返回类型。模板必须配置为SmartMessageConverter；有关更多信息，请参见名为“使用 RabbitTemplate 从消息转换”一节。

从版本 2.1 开始，可以使用noLocalReplyConsumer选项配置RabbitTemplate来控制noLocal标志以供回复使用者。现在默认为false。

Reply Timeout

默认情况下，send 和 receive 方法将在 5 秒后超时并返回 null。可以通过设置replyTimeout属性来修改。从版本 1.5 开始，如果将mandatory属性设置为 true(或对于特定消息，mandatory-expression评估为true)，则如果无法将消息传递到队列，则会抛出AmqpMessageReturnedException。此异常具有returnedMessage，replyCode，replyText属性，以及用于发送的exchange和routingKey。

从版本 2.1.2 开始，已添加方法replyTimedOut，允许将超时通知子类，以便子类可以清除任何保留的状态。

RabbitMQ 直接回复

命名队列(amq.rabbitmq.reply-to除外)仍支持回复侦听器，从而可以控制回复并发等。

从 1.6 版开始，如果由于某种原因希望对每个答复使用一个临时的，排他的，自动删除队列，请将useTemporaryReplyQueues属性设置为true。如果您设置replyAddress，则将忽略此属性。

通过将RabbitTemplate子类化并覆盖useDirectReplyTo()，可以更改是否使用直接回复的决定以使用不同的标准。该方法仅被调用一次。发送第一个请求时。

在版本 2.0 之前的版本中，RabbitTemplate为每个请求创建一个新使用者，并在收到答复(或超时)时取消该使用者。现在，模板改为使用DirectReplyToMessageListenerContainer，从而允许使用者重新使用。模板仍然负责将回复相关联，因此不会有延迟回复发送给其他发件人的危险。如果要还原为以前的行为，请将属性useDirectReplyToContainer(使用 XML 配置时为direct-reply-to-container)设置为 false。

AsyncRabbitTemplate没有这样的选项-使用直接 replyTo 时，它总是使用DirectReplyToContainer进行答复。

邮件与回复队列的相关性

使用固定的答复队列(不是amq.rabbitmq.reply-to)时，有必要提供相关数据，以便将答复与请求相关。参见RabbitMQ 远程过程调用(RPC)。默认情况下，标准correlationId属性将用于保存相关数据。但是，如果希望使用自定义属性来保存相关数据，则可以在\ 上设置correlation-key属性。将该属性显式设置为correlationId与省略该属性相同。当然，Client 端和服务器必须将相同的 Headers 用于关联数据。

默认情况下，模板将生成自己的关联 ID(忽略任何用户提供的值)。如果您希望使用自己的 correlationId，请将RabbitTemplate的userCorrelationId属性设置为true。

回复侦听器容器

当使用* 3.4.0 *之前的 RabbitMQ 版本时，每个答复都会使用一个新的临时队列。但是，可以在模板上配置单个回复队列，这样可以提高效率，并且还可以在该队列上设置参数。但是，在这种情况下，您还必须提供\ 子元素。此元素为答复队列提供一个侦听器容器，其中模板为侦听器。<listener-container/>上允许的所有第 3.1.17 节“消息侦听器容器配置”属性都在元素上允许，但从模板配置继承的 connection-factory 和 message-converter 除外。

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

尽管容器和模板共享连接工厂，但是它们不共享通道，因此不会在同一事务(如果是事务性)内执行请求和答复。

在此配置下，SimpleListenerContainer用于接收回复； RabbitTemplate是MessageListener。如上所示，当使用<rabbit:template/>名称空间元素定义模板时，解析器会将模板中的容器和连线定义为侦听器。

如果将RabbitTemplate定义为<bean/>，或者使用@Configuration类将其定义为@Bean，或者以编程方式创建模板，则需要自己定义并连接答复侦听器容器。如果您无法执行此操作，则模板将永远不会收到答复，并且最终将超时并返回空值作为对sendAndReceive方法的调用的答复。

从* version 1.5 *开始，RabbitTemplate将检测是否已将其配置为MessageListener以接收答复。否则，尝试发送和接收带有回复地址的消息将失败，并带有IllegalStateException(因为将永远不会收到回复)。

此外，如果使用简单的replyAddress(队列名称)，则回复侦听器容器将验证它正在侦听具有相同名称的队列。如果回复地址是交换和路由密钥，则将无法执行此检查，并且将写入调试日志消息。

以下是如何手动连接 bean 的示例。

<bean id="amqpTemplate" class="org.springframework.amqp.rabbit.core.RabbitTemplate">
    <constructor-arg ref="connectionFactory" />
    <property name="exchange" value="foo.exchange" />
    <property name="routingKey" value="foo" />
    <property name="replyQueue" ref="replyQ" />
    <property name="replyTimeout" value="600000" />
</bean>

<bean class="org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer">
    <constructor-arg ref="connectionFactory" />
    <property name="queues" ref="replyQ" />
    <property name="messageListener" ref="amqpTemplate" />
</bean>

<rabbit:queue id="replyQ" name="my.reply.queue" />

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

这个测试用例中显示了连接固定的答复队列的RabbitTemplate以及处理请求并返回答复的“远程”侦听器容器的完整示例。

在版本 1.3.6 *之前，仅记录了对超时消息的最新答复。现在，如果收到延迟答复，则拒绝该答复(模板抛出AmqpRejectAndDontRequeueException)。如果将答复队列配置为将拒绝的消息发送到死信交换，则可以检索答复以供以后分析。只需使用等于答复队列名称的路由密钥将队列绑定到配置的死信交换。

有关配置无效字母的更多信息，请参考RabbitMQ 死信文档。您还可以查看FixedReplyQueueDeadLetterTests测试用例作为示例。

AsyncRabbitTemplate

• 1.6 版*引入了AsyncRabbitTemplate。它具有与AmqpTemplate相似的sendAndReceive(和convertSendAndReceive)方法，但是它们不是阻塞而是返回ListenableFuture。

sendAndReceive方法返回RabbitMessageFuture; convertSendAndReceive方法返回RabbitConverterFuture。

您可以稍后在以后调用get()来同步检索结果，也可以注册将与结果异步调用的回调。

@Autowired
private AsyncRabbitTemplate template;

...

public void doSomeWorkAndGetResultLater() {

    ...

    ListenableFuture<String> future = this.template.convertSendAndReceive("foo");

    // do some more work

    String reply = null;
    try {
        reply = future.get();
    }
    catch (ExecutionException e) {
        ...
    }

    ...

}

public void doSomeWorkAndGetResultAsync() {

    ...

    RabbitConverterFuture<String> future = this.template.convertSendAndReceive("foo");
    future.addCallback(new ListenableFutureCallback<String>() {

        @Override
        public void onSuccess(String result) {
            ...
        }

        @Override
        public void onFailure(Throwable ex) {
            ...
        }

    });

    ...

}

如果设置了mandatory，并且消息无法传递，则将来会抛出ExecutionException，原因为AmqpMessageReturnedException，该原因封装了返回的消息和有关返回的信息。

如果设置了enableConfirms，则将来将具有属性confirm，该属性本身是ListenableFuture<Boolean>，其中true表示发布成功。如果确认 Future 为假，则RabbitFuture将具有另一个属性nackCause-失败的原因(如果有)。

在模板上设置receiveTimeout属性以使答复超时(默认为30000-30 秒)。如果发生超时，将来将以AmqpReplyTimeoutException结束。

模板实现SmartLifecycle;在有待处理的回复时停止模板将导致待处理的Future s 被取消。

从* version 2.0 *版本开始，异步模板现在支持Direct reply-to而不是配置的回复队列。要启用此功能，请使用以下构造函数之一：

public AsyncRabbitTemplate(ConnectionFactory connectionFactory, String exchange, String routingKey)

public AsyncRabbitTemplate(RabbitTemplate template)

请参阅“ RabbitMQ 直接回复”部分以与同步RabbitTemplate一起使用直接回复。

从* version 2.0 *开始，这些方法(convertSendAndReceiveAsType)的变体需要一个附加的ParameterizedTypeReference参数来转换复杂的返回类型。基础RabbitTemplate必须配置为SmartMessageConverter；有关更多信息，请参见名为“使用 RabbitTemplate 从消息转换”一节。

使用 AMQP 进行远程处理

Spring 框架具有一般的远程处理功能，允许远程过程调用(RPC)使用各种传输方式。 Spring-AMQP 支持类似的机制，Client 端上为AmqpProxyFactoryBean，服务器上为AmqpInvokerServiceExporter。这提供了基于 AMQP 的 RPC。在 Client 端，如上所述使用RabbitTemplate；在服务器端，调用程序(配置为MessageListener)接收消息，调用配置的服务，并使用入站消息的replyTo信息返回答复。

可以将 Client 端工厂 bean 注入到任何 bean 中(使用其serviceInterface)；然后，Client 端可以调用代理上的方法，从而通过 AMQP 进行远程执行。

在服务器端，AmqpInvokerServiceExporter同时具有AmqpTemplate和MessageConverter属性。当前，未使用模板的MessageConverter。如果需要提供自定义消息转换器，则应使用messageConverter属性提供它。在 Client 端，可以将自定义消息转换器添加到AmqpTemplate，使用其amqpTemplate属性将其提供给AmqpProxyFactoryBean。

Client 端和服务器配置示例如下所示。

<bean id="client"
	class="org.springframework.amqp.remoting.client.AmqpProxyFactoryBean">
	<property name="amqpTemplate" ref="template" />
	<property name="serviceInterface" value="foo.ServiceInterface" />
</bean>

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

<rabbit:template id="template" connection-factory="connectionFactory" reply-timeout="2000"
	routing-key="remoting.binding" exchange="remoting.exchange" />

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

<rabbit:queue name="remoting.queue" />

<rabbit:direct-exchange name="remoting.exchange">
	<rabbit:bindings>
		<rabbit:binding queue="remoting.queue" key="remoting.binding" />
	</rabbit:bindings>
</rabbit:direct-exchange>

<bean id="listener"
	class="org.springframework.amqp.remoting.service.AmqpInvokerServiceExporter">
	<property name="serviceInterface" value="foo.ServiceInterface" />
	<property name="service" ref="service" />
	<property name="amqpTemplate" ref="template" />
</bean>

<bean id="service" class="foo.ServiceImpl" />

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

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

<rabbit:queue name="remoting.queue" />

<rabbit:listener-container connection-factory="connectionFactory">
	<rabbit:listener ref="listener" queue-names="remoting.queue" />
</rabbit:listener-container>

3.1.11 配置代理

Introduction

AMQP 规范描述了如何使用协议在代理上配置队列，交换和绑定。这些可从 0.8 规范及更高版本移植的操作位于org.springframework.amqp.core软件包的AmqpAdmin界面中。该类的 RabbitMQ 实现位于org.springframework.amqp.rabbit.core包中的RabbitAdmin。

AmqpAdmin 界面基于使用 Spring AMQP 域抽象的基础，如下所示：

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);

}

另请参见称为“范围操作”的部分。

getQueueProperties()方法返回有关队列的一些有限信息(消息计数和使用者计数)。返回的属性的键在RabbitTemplate(QUEUE_NAME，QUEUE_MESSAGE_COUNT，QUEUE_CONSUMER_COUNT)中可用作常量。 RabbitMQ REST API在QueueInfo对象中提供了更多信息。

no-arg declareQueue()方法使用自动生成的名称在代理上定义队列。此自动生成的队列的其他属性是exclusive=true，autoDelete=true和durable=false。

declareQueue(Queue queue)方法采用Queue对象，并返回声明的队列的名称。如果提供的Queue的name属性为空 String，则代理将使用生成的名称声明队列，并将该名称返回给调用方。该名称也被添加到Queue的actualName属性中。该功能只能通过直接调用RabbitAdmin以编程方式使用。当 Management 员通过在应用程序上下文中声明性地定义队列来使用自动声明时，请将 name 属性设置为""，然后代理将创建该名称。从* version 2.1 *开始，侦听器容器可以使用这种类型的队列。有关更多信息，请参见第 3.1.7 节“容器和代理命名的队列”。

这与AnonymousQueue相反，在AnonymousQueue中，框架生成唯一的(UUID)名称，并将durable设置为false和exclusive，autoDelete设置为true。具有name空属性或缺少name属性的<rabbit:queue/>将始终创建AnonymousQueue。

请参阅名为“ AnonymousQueue”的部分，以了解为什么AnonymousQueue优先于代理生成的队列名称，以及如何控制名称的格式。从 2.1 版开始，默认情况下，匿名队列声明的参数x-queue-master-locator设置为client-local。这样可以确保在应用程序连接到的节点上声明队列。声明性队列必须具有固定的名称，因为它们可能在上下文中的其他地方(例如，在侦听器中)被引用：

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

See 称为“交换，队列和绑定的自动声明”的部分.

该接口的 RabbitMQ 实现是 RabbitAdmin，当使用 Spring XML 进行配置时，它将如下所示：

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

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

当CachingConnectionFactory缓存模式为CHANNEL(默认设置)时，RabbitAdmin实现会自动在同一ApplicationContext中声明的Queues，Exchanges和Bindings的惰性声明。向代理打开Connection后，将立即声明这些组件。有一些名称空间功能使此操作非常方便，例如在“股票”samples 应用程序中，我们具有：

<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 引用它们。我们还可以使用显式名称声明 Queue，这些名称也用作上下文中其 bean 定义的标识符。例如。

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

可以使用其他参数配置队列，例如x-message-ttl。使用名称空间支持，它们使用<rabbit:queue-arguments>元素以实参名称/实参值对的 Map 的形式提供。

<rabbit:queue name="withArguments">
    <rabbit:queue-arguments>
        <entry key="x-dead-letter-exchange" value="myDLX"/>
        <entry key="x-dead-letter-routing-key" value="dlqRK"/>
    </rabbit:queue-arguments>
</rabbit:queue>

默认情况下，参数被假定为字符串。对于其他类型的参数，需要提供类型。

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

提供混合类型的参数时，将为每个入口元素提供类型：

<rabbit:queue name="withArguments">
    <rabbit:queue-arguments>
        <entry key="x-message-ttl">
            <value type="java.lang.Long">100</value>
        </entry>
        <entry key="x-dead-letter-exchange" value="myDLX"/>
        <entry key="x-dead-letter-routing-key" value="dlqRK"/>
    </rabbit:queue-arguments>
</rabbit:queue>

在 Spring Framework 3.2 和更高版本中，这可以更简洁地声明：

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

使用 Java 配置时，通过Queue类上的setMasterLocator()方法将x-queue-master-locator作为第一类属性来支持。从 2.1 版开始，默认情况下，匿名队列声明此属性设置为client-local。这样可以确保在应用程序连接到的节点上声明队列。

默认情况下，发生任何异常时，RabbitAdmin将立即停止处理所有声明。这可能会导致下游问题-例如“监听器容器”初始化失败，因为未声明另一个队列(在错误的队列之后定义)。

可以通过将RabbitAdmin上的ignore-declaration-exceptions属性设置为true来修改此行为。此选项指示RabbitAdmin记录异常，并 continue 声明其他元素。使用 Java 配置RabbitAdmin时，此属性为ignoreDeclarationExceptions。这是适用于所有元素的全局设置，队列，交换和绑定具有类似的属性，仅适用于那些元素。

在*版本 1.6 *之前，仅当通道上出现IOException时(例如，当前属性与所需属性之间不匹配时)，此属性才生效。现在，此属性对任何异常(包括TimeoutException等)都生效。

此外，任何声明异常都会导致发布DeclarationExceptionEvent，这是一个ApplicationEvent，上下文中的任何ApplicationListener都可以使用它。该事件包含对 admin，正在声明的元素和Throwable的引用。

Headers Exchange

从*版本 1.3 *开始，可以将HeadersExchange配置为在多个 Headers 上匹配；您还可以指定是否必须匹配任何 Headers：

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

从* version 1.6 *开始，Exchanges可以配置为internal标志(默认为false)，并且可以通过RabbitAdmin在 Broker 上正确配置这样的Exchange(如果应用程序上下文中存在)。如果用于交换的internal标志是true，RabbitMQ 将不允许 Client 端使用交换。这对于无用信函交换或交换到交换绑定非常有用，在这种情况下，您不希望发行人直接使用交换。

要查看如何使用 Java 配置 AMQP 基础结构，请查看 Stock 示例应用程序，其中有@Configuration类AbstractStockRabbitConfiguration，而该类又具有RabbitClientConfiguration和RabbitServerConfiguration子类。 AbstractStockRabbitConfiguration的代码如下所示

@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 Jackson2JsonMessageConverter();
    }

    @Bean
    public TopicExchange marketDataExchange() {
        return new TopicExchange("app.stock.marketdata");
    }

    // additional code omitted for brevity

}

在 Stock 应用程序中，使用以下@Configuration 类配置服务器：

@Configuration
public class RabbitServerConfiguration extends AbstractStockAppRabbitConfiguration  {

    @Bean
    public Queue stockRequestQueue() {
        return new Queue("app.stock.request");
    }
}

这是@Configuration类的整个继承链的结尾。最终结果是在应用程序启动时将TopicExchange和Queue声明给代理。在服务器配置中，没有将 TopicExchange 绑定到队列，这是在 Client 端应用程序中完成的。但是，库存请求队列自动绑定到 AMQP 默认交换-此行为由规范定义。

Client 端@Configuration类更加有趣，如下所示。

@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

}

Client 端正在通过AmqpAdmin上的declareQueue()方法声明另一个队列，并且该 Client 端使用在属性文件中外部化的路由模式将该队列绑定到市场数据交换。

用于队列和交换的 Builder API

*版本 1.6 *引入了便捷的 fluent API，用于在使用 Java 配置时配置Queue和Exchange对象：

@Bean
public Queue queue() {
    return QueueBuilder.nonDurable("foo")
        .autoDelete()
        .exclusive()
        .withArgument("foo", "bar")
        .build();
}

@Bean
public Exchange exchange() {
  return ExchangeBuilder.directExchange("foo")
      .autoDelete()
      .internal()
      .withArgument("foo", "bar")
      .build();
}

有关更多信息，请参见org.springframework.amqp.core.QueueBuilder和org.springframework.amqp.core.ExchangeBuilder的 javadocs。

从* version 2.0 *版本开始，ExchangeBuilder现在默认情况下会创建持久交换，以与各个AbstractExchange类上的简单构造函数保持一致。要与构建器进行非持久交换，请在调用.build()之前使用.durable(false)。不再提供没有参数的durable()方法。

声明交换，队列和绑定的集合

您可以在Declarables个对象中包装可声明对象(Queue，Exchange，Binding)的集合。 RabbitAdmin将在应用程序上下文中检测到此类 bean，并在代理上声明所包含的对象。

@Configuration
public static class Config {

    @Bean
    public ConnectionFactory cf() {
        return new CachingConnectionFactory("localhost");
    }

    @Bean
    public RabbitAdmin admin(ConnectionFactory cf) {
        return new RabbitAdmin(cf);
    }

    @Bean
    public DirectExchange e1() {
    	return new DirectExchange("e1", false, true);
    }

    @Bean
    public Queue q1() {
    	return new Queue("q1", false, false, true);
    }

    @Bean
    public Binding b1() {
    	return BindingBuilder.bind(q1()).to(e1()).with("k1");
    }

    @Bean
    public Declarables es() {
        return new Declarables(
                new DirectExchange("e2", false, true),
                new DirectExchange("e3", false, true));
    }

    @Bean
    public Declarables qs() {
        return new Declarables(
                new Queue("q2", false, false, true),
                new Queue("q3", false, false, true));
    }

    @Bean
    @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
    public Declarables prototypes() {
        return new Declarables(new Queue(this.prototypeQueueName, false, false, true));
    }

    @Bean
    public Declarables bs() {
        return new Declarables(
                new Binding("q2", DestinationType.QUEUE, "e2", "k2", null),
                new Binding("q3", DestinationType.QUEUE, "e3", "k3", null));
    }

    @Bean
    public Declarables ds() {
        return new Declarables(
                new DirectExchange("e4", false, true),
                new Queue("q4", false, false, true),
                new Binding("q4", DestinationType.QUEUE, "e4", "k4", null));
    }

}

Conditional Declaration

默认情况下，所有队列，交换和绑定都由应用程序上下文中的所有RabbitAdmin个实例(具有auto-startup="true")声明。

表示这些元素的类实现Declarable，它具有两种方法：shouldDeclare()和getDeclaringAdmins()。 RabbitAdmin使用这些方法来确定特定实例是否应实际处理其Connection上的声明。

这些属性可用作名称空间中的属性，如以下示例所示。

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

同样，您可以使用基于 Java 的@Configuration来达到相同的效果。在此示例中，组件将由admin1而不是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;
}

关于“ id”和“ name”属性的 Comments

<rabbit:queue/>和<rabbit:exchange/>元素上的name属性反映代理中实体的名称。对于队列，如果省略name，则会创建一个匿名队列(请参见下面的名为“ AnonymousQueue”的部分)。

在 2.0 之前的版本中，name也已注册为 Bean 名称别名(类似于<bean/>元素上的name)。

这导致了两个问题：

• 它阻止了队列的声明和同名交换

• 如果别名包含 SpEL 表达式(#{...})，则无法解析

从* version 2.0 版本开始，如果您同时使用id和 name属性声明这些元素之一，则该名称将不再声明为 Bean 名称别名。如果要声明队列并与相同的name交换，则必须提供id。

如果该元素仅具有name属性，则不会发生任何变化；例如，在绑定声明中，该 Bean 仍可以被name引用，但是如果名称包含 SpEL，则仍然无法引用它-请提供id以供参考。

AnonymousQueue

通常，当需要唯一命名的，排他的自动删除队列时，建议使用AnonymousQueue代替代理定义的队列名称(使用""作为Queue名称将导致代理生成队列名称)。

这是因为：

• 构建与代理的连接时，实际上会声明队列。这些 bean 创建并连接在一起很长时间之后；使用队列的 bean 需要知道其名称。实际上，启动应用程序时，代理甚至可能没有运行。

• 如果由于某种原因失去了与代理的连接，则 Management 员将重新声明具有相同名称的AnonymousQueue。如果使用代理声明的队列，则队列名称将更改。

您可以控制AnonymousQueue s 使用的队列名称的格式。

默认情况下，队列名称以spring.gen-为前缀，后跟UUID的 base64 表示形式，例如spring.gen-MRBv9sqISkuCiPfOYfpo4g。

您可以在构造函数参数中提供AnonymousQueue.NamingStrategy实现。

@Bean
public Queue anon1() {
    return new AnonymousQueue();
}

@Bean
public Queue anon2() {
    return new AnonymousQueue(new AnonymousQueue.Base64UrlNamingStrategy("foo-"));
}

@Bean
public Queue anon3() {
    return new AnonymousQueue(AnonymousQueue.UUIDNamingStrategy.DEFAULT);
}

第一个将生成以spring.gen-为前缀的队列名称，后跟UUID的 base64 表示形式，例如spring.gen-MRBv9sqISkuCiPfOYfpo4g。第二个将生成以foo-为前缀的队列名称，后跟UUID的 base64 表示形式。第三个将仅使用 UUID 生成名称(不进行 base64 转换)，例如f20c818a-006b-4416-bf91-643590fedb0e。

base64 编码使用 RFC 4648 中的“ URL 和文件名安全字母”；尾随的填充字符(=)被删除。

您可以提供自己的命名策略，从而可以在队列名称中包含其他信息(例如，应用程序，Client 端主机)。

使用 XML 配置时可以指定命名策略。 naming-strategy属性存在于实现AnonymousQueue.NamingStrategy的 Bean 引用的<rabbit:queue>元素上。

<rabbit:queue id="uuidAnon" />

<rabbit:queue id="springAnon" naming-strategy="uuidNamer" />

<rabbit:queue id="customAnon" naming-strategy="customNamer" />

<bean id="uuidNamer" class="org.springframework.amqp.core.AnonymousQueue.UUIDNamingStrategy" />

<bean id="customNamer" class="org.springframework.amqp.core.AnonymousQueue.Base64UrlNamingStrategy">
    <constructor-arg value="custom.gen-" />
</bean>

第一个创建的名称类似于spring.gen-MRBv9sqISkuCiPfOYfpo4g。第二个方法使用 UUID 的字符串表示形式创建名称。第三个创建类似custom.gen-MRBv9sqISkuCiPfOYfpo4g的名称。

当然，您可以提供自己的命名策略 bean。

从 2.1 版开始，默认情况下将使用参数x-queue-master-locator设置为client-local声明匿名队列。这样可以确保在应用程序连接到的节点上声明队列。若要还原为以前的行为，请在构造实例后调用queue.setMasterLocator(null)。

3.1.12broker 事件监听器

启用事件交换插件时，如果将类型BrokerEventListener的 bean 添加到应用程序上下文中，它将以BrokerEvent的形式发布选定的代理事件，可以通过常规 Spring ApplicationListener或@EventListener方法使用。代理将事件发布到主题交换amq.rabbitmq.event，每个事件类型使用不同的路由键。具有事件键的侦听器，用于将AnonymousQueue绑定到交换机，因此侦听器将仅接收选定的事件；由于这是主题交换，因此可以使用通配符(以及显式请求特定事件)

@Bean
public BrokerEventListener eventListener() {
    return new BrokerEventListener(connectionFactory(), "user.deleted", "channel.#", "queue.#");
}

您可以使用常规的 Spring 技术进一步缩小单个事件侦听器中接收到的事件的范围：

@EventListener(condition = "event.eventType == 'queue.created'")
public void listener(BrokerEvent event) {
    ...
}

3.1.13 邮件交换延迟

• 1.6 版*引入了对延迟的消息交换插件的支持

要使用RabbitAdmin声明延迟交换，只需将交换 bean 上的delayed属性设置为 true。 RabbitAdmin将使用交换类型(Direct，Fanout等)来设置x-delayed-type参数，并使用x-delayed-message类型声明交换。

使用 XML 配置交换 bean 时，delayed属性(默认false)也可用。

<rabbit:topic-exchange name="topic" delayed="true" />

要发送延迟的消息，只需通过MessageProperties设置x-delayHeaders 即可：

MessageProperties properties = new MessageProperties();
properties.setDelay(15000);
template.send(exchange, routingKey,
        MessageBuilder.withBody("foo".getBytes()).andProperties(properties).build());

or

rabbitTemplate.convertAndSend(exchange, routingKey, "foo", new MessagePostProcessor() {

    @Override
    public Message postProcessMessage(Message message) throws AmqpException {
        message.getMessageProperties().setDelay(15000);
        return message;
    }

});

要检查消息是否延迟，请使用MessageProperties上的getReceivedDelay()方法。这是一个单独的属性，可以避免意外传播到由 Importing 消息生成的输出消息。

3.1.14 RabbitMQ REST API

启用 Management 插件后，RabbitMQ 服务器将公开 REST API 来监视和配置代理。现在提供了API 的 Java 绑定。 com.rabbitmq.http.client.Client是标准的，即时的且因此是阻塞的 API。它基于Spring Web模块及其RestTemplate实现。另一方面，com.rabbitmq.http.client.ReactorNettyClient是基于Reactor Netty项目的 Reactive，非阻塞实现。

跳相关性(com.rabbitmq:http-client)现在也为optional。

有关更多信息，请参考其 javadocs。

3.1.15 异常处理

RabbitMQ JavaClient 端的许多操作都可能引发已检查的异常。例如，在很多情况下，可能会引发 IOExceptions。 RabbitTemplate，SimpleMessageListenerContainer 和其他 Spring AMQP 组件将捕获这些 Exception，并将其转换为我们的运行时层次结构中的 Exception 之一。这些在* org.springframework.amqp *包中定义，并且 AmqpException 是层次结构的基础。

当侦听器引发异常时，它将包裹在ListenerExecutionFailedException中，通常，消息将被代理拒绝并重新排队。将defaultRequeueRejected设置为 false 将导致消息被丢弃(或路由到无效信件交换)。如称为“消息侦听器和异步情况”的部分中所述，侦听器可以抛出AmqpRejectAndDontRequeueException(或ImmediateRequeueAmqpException)以有条件地控制此行为。

但是，存在一类错误，侦听器无法控制该行为。当遇到无法转换的消息(例如无效的content_encoding头)时，在消息到达用户代码之前会引发一些异常。将defaultRequeueRejected设置为true(默认值)(或抛出ImmediateRequeueAmqpException)，此类消息将一遍又一遍地传递。在*版本 1.3.2 *之前，用户需要编写自定义ErrorHandler，如第 3.1.15 节“异常处理”中所述，以避免这种情况。

从* version 1.3.2 *开始，默认的ErrorHandler现在是ConditionalRejectingErrorHandler，它将拒绝(而不是重新排队)由于不可恢复的错误而失败的消息：

• o.s.amqp...MessageConversionException

• o.s.messaging...MessageConversionException

• o.s.messaging...MethodArgumentNotValidException

• o.s.messaging...MethodArgumentTypeMismatchException

• java.lang.NoSuchMethodException

• java.lang.ClassCastException

使用MessageConverter转换传入消息有效负载时，可以引发第一个。如果 Map 到@RabbitListener方法时需要其他转换，则转换服务可能会抛出第二个错误。如果在侦听器中使用了验证(例如@Valid)并且验证失败，则可能会抛出第三个。如果将入站消息转换为对目标方法不正确的类型，则可能会抛出第四个错误。例如，将参数声明为Message<Foo>，但接收到Message<Bar>。

在*版本 1.6.3 *中添加了第五和第六。

可以使用FatalExceptionStrategy配置此错误处理程序的实例，以便用户可以提供自己的条件消息拒绝规则，例如从 Spring Retry(称为“消息侦听器和异步情况”的部分)到BinaryExceptionClassifier的委托实现。此外，ListenerExecutionFailedException现在具有failedMessage属性，可以在决策中使用该属性。如果FatalExceptionStrategy.isFatal()方法返回true，则错误处理程序将抛出AmqpRejectAndDontRequeueException。当确定异常是致命的时，默认值FatalExceptionStrategy会记录一条警告消息。

从 1.6.3 版开始，将用户异常添加到致命列表的便捷方法是子类ConditionalRejectingErrorHandler.DefaultExceptionStrategy并重写方法isUserCauseFatal(Throwable cause)以针对致命异常返回 true。

处理 DLQ 消息的常见模式是在这些消息上设置time-to-live以及其他 DLQ 配置，以使这些消息过期并路由回主队列以重试。这种技术的问题在于，导致致命异常的消息将永远循环。从版本 2.1 开始，ConditionalRejectingErrorHandler在邮件上检测到x-deathHeaders，该 Headers 导致引发致命异常，该邮件将被记录并丢弃。

3.1.16 Transactions

Introduction

Spring Rabbit 框架支持同步和异步用例中的自动事务 Management，具有许多不同的语义，可以pass 语句方式选择这些语义，这是 Spring 事务的现有用户所熟悉的。这使许多(如果不是大多数)常见的消息传递模式非常易于实现。

有两种方法可以向框架发出所需的事务语义。在RabbitTemplate和SimpleMessageListenerContainer中都有一个标志channelTransacted，如果为 true，则指示框架使用事务通道并根据结果以提交或回滚来结束所有操作(发送或接收)，并带有表示回滚的异常。 。另一个 signal 是使用 Spring 的PlatformTransactionManager实现之一提供外部事务，作为正在进行的操作的上下文。如果在框架发送或接收消息时已经有一个事务在进行中，并且channelTransacted标志为 true，则将消息事务的提交或回滚推迟到当前事务结束为止。如果channelTransacted标志为 false，则没有事务语义适用于消息传递操作(它是自动确认的)。

channelTransacted标志是配置时间设置：创建 AMQP 组件时，通常在应用程序启动时声明并处理一次。原则上，外部事务是动态的，因为系统会在运行时响应当前的线程状态，但实际上，当将事务声明性地分层到应用程序时，外部事务通常也是配置设置。

对于带有RabbitTemplate的同步用例，外部事务由调用方根据喜好以声明性或强制性方式提供(通常的 Spring 事务模型)。声明性方法的示例(通常首选，因为它是非侵入性的)，其中模板已使用channelTransacted=true配置：

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

在标记为@Transactional 的方法内，将 String 有效负载作为消息主体进行接收，转换和发送，因此，如果数据库处理因异常而失败，则传入消息将返回给代理，而传出消息将不会被发送。这适用于在 Transaction 方法链中使用RabbitTemplate进行的任何操作(例如，除非直接操纵Channel以尽早提交 Transaction)。

对于带有SimpleMessageListenerContainer的异步用例，如果需要外部事务，则容器在设置侦听器时必须请求它。为了表明需要进行外部 Transaction，用户在配置容器后向其提供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;
    }

}

在上面的示例中，添加了事务 Management 器作为从另一个 bean 定义(未显示)注入的依赖项，并且channelTransacted标志也设置为 true。结果是，如果侦听器因异常而失败，则事务将回滚，并且消息也将返回给代理。重要的是，如果事务提交失败(例如，数据库约束错误或连接问题)，则 AMQP 事务也将回滚，并将消息返回给代理。这有时被称为“尽力而为第一阶段提交”，是可靠消息传递的一种非常强大的模式。如果在上面的示例中将channelTransacted标志设置为 false(这是默认设置)，则仍将为侦听器提供外部事务，但是所有消息传递操作将被自动确认，因此结果是即使提交消息传递操作也是如此在业务操作上回滚。

Conditional Rollback

在版本 1.6.6 *之前，使用外部事务 Management 器(例如 JDBC)时，将回滚规则添加到容器的transactionAttribute无效；异常总是回滚事务。

另外，在容器的通知链中使用transaction advice时，条件回滚不是很有用，因为所有侦听器异常都包装在ListenerExecutionFailedException中。

第一个问题已得到纠正，现在可以正确应用规则。此外，现在提供了ListenerFailedRuleBasedTransactionAttribute；它是RuleBasedTransactionAttribute的子类，唯一的区别是它知道ListenerExecutionFailedException并将该异常的原因用于规则。该 Transaction 属性可以直接在容器中使用，也可以通过 Transaction 通知使用。

使用此规则的示例如下：

@Bean
public AbstractMessageListenerContainer container() {
    ...
    container.setTransactionManager(transactionManager);
    RuleBasedTransactionAttribute transactionAttribute =
        new ListenerFailedRuleBasedTransactionAttribute();
    transactionAttribute.setRollbackRules(Collections.singletonList(
        new NoRollbackRuleAttribute(DontRollBackException.class)));
    container.setTransactionAttribute(transactionAttribute);
    ...
}

关于回滚收到的邮件的 Comments

AMQP 事务仅适用于发送给代理的消息和确认，因此，在对 Spring 事务进行回滚并且已接收到消息时，Spring AMQP 要做的不仅是回滚事务，而且还手动拒绝消息(有点小问题，但这不是规范所称的)。对消息拒绝采取的操作与事务无关，并且取决于defaultRequeueRejected属性(默认为true)。有关拒绝失败消息的更多信息，请参阅称为“消息侦听器和异步情况”的部分。

有关 RabbitMQ 事务及其限制的更多信息，请参阅RabbitMQbroker 语义。

使用 RabbitTransactionManager

RabbitTransactionManager是在外部事务内执行并与外部事务同步的 Rabbit 操作的替代方法。该事务 Management 器是PlatformTransactionManager接口的实现，应与单个 Rabbit ConnectionFactory 一起使用。

需要应用程序代码来通过ConnectionFactoryUtils.getTransactionalResourceHolder(ConnectionFactory, boolean)而不是随后的 Channel 创建一个标准Connection.createChannel()来通过ConnectionFactoryUtils.getTransactionalResourceHolder(ConnectionFactory, boolean)检索事务型 Rabbit 资源。使用 Spring AMQP 的RabbitTemplate时，它将自动检测线程绑定的 Channel 并自动参与其事务。

使用 Java 配置，您可以使用以下命令设置新的 RabbitTransactionManager：

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

如果您更喜欢使用 XML 配置，请在 XML Application Context 文件中声明以下 bean：

<bean id="rabbitTxManager"
      class="org.springframework.amqp.rabbit.transaction.RabbitTransactionManager">
    <property name="connectionFactory" ref="connectionFactory"/>
</bean>

3.1.17 邮件侦听器容器配置

用于配置与事务和服务质量相关的SimpleMessageListenerContainer(SMLC)和DirectMessageListenerContainer(DMLC)的选项很多，其中一些彼此交互。适用于 SMLC 或 DMLC 的属性由相应列中的复选标记指示。请参阅名为“选择容器”的部分，以获取帮助您确定适合您的应用程序的容器的信息。

下表显示了使用命名空间配置<rabbit:listener-container/>时的容器属性名称及其等效属性名称(在括号中)。该元素上的type属性可以是simple(默认值)或direct以分别指定SMLC或DMLC。

命名空间未公开某些属性。由N/A表示的属性。

表 3.3. 消息侦听器容器的配置选项

Property (Attribute)	Description	SMLC	DMLC
(group)	仅在使用名称空间时才可用。指定后，将使用该名称注册类型为Collection<MessageListenerContainer>的 bean，并将每个<listener/>元素的容器添加到集合中。例如，这允许通过遍历集合来启动/停止容器组。如果多个<listener-container/>元素具有相同的组值，则集合中的容器是所有指定容器的集合。
channelTransacted
(channel-transacted)	布尔值标志，表示应在事务中(手动或自动)确认所有消息
acknowledgeMode (确认)	NONE =将不发送任何 signal(与channelTransacted=true不兼容)。 RabbitMQ 之所以称其为“自动确认”，是因为代理假定所有消息都已确认，而无需使用者采取任何措施。 MANUAL =侦听器必须通过调用Channel.basicAck()来确认所有消息。 AUTO =容器将自动确认该消息，除非MessageListener引发异常。请注意，acknowledgeMode是 channelTransacted 的补充-如果处理了通道，则代理除了确认外还需要提交通知。这是默认模式。另请参见txSize。
transactionManager (事务 Management 器)	用于侦听器操作的外部事务 Management 器。也是对 channelTransacted 的补充-如果处理Channel，则其事务将与外部事务同步。
prefetchCount (预取)	每个使用者可能未完成的未确认消息数。越高，消息传递的速度越快，但是非 Sequences 处理的风险也越高。忽略acknowledgeMode是否为 NONE。如有必要，它将增加以匹配txSize或messagePerAck。从 2.0 开始默认为 250；设置为 1 可恢复以前的行为。

3.1.18 侦听器并发

SimpleMessageListenerContainer

默认情况下，侦听器容器将启动一个使用者，该使用者将从队列中接收消息。

在检查上一节中的表时，您将看到许多控制并发性的属性/属性。最简单的是concurrentConsumers，它仅创建(固定)数量的使用者，该使用者将同时处理消息。

在*版本 1.3.0 *之前，这是唯一可用的设置，必须停止容器并再次启动容器才能更改设置。

从*版本 1.3.0 *开始，您现在可以动态调整concurrentConsumers属性。如果在容器运行时更改它，则将根据需要添加或删除使用者，以适应新设置。

此外，已添加新属性maxConcurrentConsumers，并且容器将根据工作负载动态调整并发性。这与四个附加属性一起工作：consecutiveActiveTrigger，startConsumerMinInterval，consecutiveIdleTrigger，stopConsumerMinInterval。使用默认设置，增加使用方的算法如下：

如果尚未达到maxConcurrentConsumers且现有使用者已连续 10 个周期处于活动状态，并且自启动最后一个使用者以来至少经过了 10 秒钟，则将启动新使用者。如果 Consumer 在txSize * receiveTimeout毫秒内收到至少一条消息，则认为该 Consumer 处于活动状态。

在默认设置下，减少使用方的算法如下：

如果运行的concurrentConsumers个以上，并且使用方检测到 10 个连续超时(空闲)，并且最后一个使用方在至少 60 秒前停止，则使用方将被停止。超时取决于receiveTimeout和txSize属性。如果使用者在txSize * receiveTimeout毫秒内未收到任何消息，则认为该使用者处于空闲状态。因此，在默认超时(1 秒)和txSize为 4 的情况下，将在 40 秒的空闲时间后考虑停止使用程序(4 个超时对应于 1 个空闲检测)。

每个使用者都使用一个通道，而不管配置的队列数量如何。

从版本* 2.0 *开始，可以使用单个属性concurrency设置concurrentConsumers和maxConcurrentConsumers属性；例如"2-4"。

DirectMessageListenerContainer

使用此容器，并发性基于配置的队列和consumersPerQueue。每个队列的每个使用者都使用单独的通道，并发性由 RabbitClient 端库控制；默认情况下，它使用 5 个线程的池；您可以配置taskExecutor以提供所需的最大并发性。

3.1.19 独家 Consumer

同样从* version 1.3 *开始，可以为侦听器容器配置一个单独的使用者。这样可以防止其他容器从队列中使用，直到当前使用方被取消为止。这样的容器的并发必须为 1.

使用独占使用者时，其他容器将根据recoveryInterval属性尝试从队列中进行消耗，如果尝试失败，则记录警告。

3.1.20 侦听器容器队列

*版本 1.3 *引入了许多改进，以处理侦听器容器中的多个队列。

必须将容器配置为侦听至少一个队列。以前也是如此，但是现在可以在运行时添加和删除队列。处理完任何预提取的消息后，容器将回收(取消并重新创建)使用者。请参见方法addQueues，addQueueNames，removeQueues和removeQueueNames。删除队列时，必须至少保留一个队列。

使用者现在可以在任何队列可用的情况下启动-以前，如果任何队列不可用，容器将停止。现在，只有在没有可用队列的情况下才是这种情况。如果并非所有队列都可用，则容器将尝试每 60 秒被动声明(并消耗)丢失的队列。

同样，如果使用者从代理收到取消通知(例如，如果删除队列)，则该使用者将尝试恢复，并且恢复的使用者将 continue 处理来自任何其他已配置队列的消息。以前，取消一个队列会取消整个使用方，最终，由于缺少队列，容器将停止。

如果希望永久删除队列，则应在删除队列之前或之后更新容器，以避免将来尝试从中使用该容器。

3.1.21 弹性：从错误和代理故障中恢复

Introduction

Spring AMQP 提供的一些关键(也是最流行的)高级功能与协议错误或代理失败时的恢复和自动重新连接有关。我们已经在本指南中看到了所有相关组件，但是应该有助于将它们放在一起，并分别指出功能和恢复方案。

主要重新连接功能由CachingConnectionFactory本身启用。使用RabbitAdmin自动声明功能通常也很有益。另外，如果您关心保证交付，则可能还需要在RabbitTemplate和SimpleMessageListenerContainer中使用channelTransacted标志，在SimpleMessageListenerContainer中也使用AcknowledgeMode.AUTO(如果是您自己动手，也可以使用手册)。

自动声明交换，队列和绑定

RabbitAdmin组件可以在启动时声明交换，队列和绑定。它通过ConnectionListener懒惰地执行此操作，因此，如果启动时不存在代理，则无关紧要。第一次使用Connection(例如，通过发送消息)时，监听器将触发并应用 Management 功能。在侦听器中执行自动声明的另一个好处是，如果由于任何原因(例如代理死亡，网络故障等)而断开连接，则在重新构建连接时将再次应用它们。

另请参见“ RabbitMQ 自动连接/拓扑恢复”部分。

同步操作失败和重试选项

如果您使用RabbitTemplate(例如)以同步 Sequences 失去与代理的连接，那么 Spring AMQP 将抛出AmqpException(通常但并非总是AmqpIOException)。我们不会试图掩盖存在问题的事实，因此您必须能够捕获并响应异常。如果您怀疑连接丢失并且不是您的错，那么最简单的操作就是再次尝试该操作。您可以手动执行此操作，也可以查看使用 Spring Retry 处理(命令式或声明式)重试。

Spring Retry 提供了几个 AOP 拦截器，并提供了很大的灵 Active 来指定重试的参数(尝试次数，异常类型，退避算法等)。 Spring AMQP 还为 AMQP 用例提供了一种方便的工厂 bean，以方便的形式创建 Spring Retry 拦截器，并带有强类型的回调接口，可用于实现自定义恢复逻辑。有关更多详细信息，请参见StatefulRetryOperationsInterceptor和StatelessRetryOperationsInterceptor的 Javadocs 和属性。如果没有事务或在重试回调中启动了事务，则 Stateless 重试是合适的。请注意，Stateless 重试比有状态重试更易于配置和分析，但是如果存在正在进行的事务必须回滚或肯定要回滚，则通常不适合使用。在事务中间断开连接应该具有与回滚相同的效果，因此对于重新连接(在较高的堆栈上启动事务开始)，有状态重试通常是最佳选择。有状态重试需要一种机制来唯一标识消息。最简单的方法是让发送方在MessageId message 属性中放置一个唯一值。提供的消息转换器提供了执行此操作的选项-将createMessageIds设置为true。否则，您可以将MessageKeyGenerator实现注入到拦截器中；否则，可以将MessageKeyGenerator实现注入到拦截器中。密钥生成器必须为每个消息返回唯一的密钥。在* version 2.0 *之前的版本中，提供了MissingMessageIdAdvice，使没有messageId属性的消息仅被重试一次(忽略重试设置)。由于与spring-retry 1.2 版一起将其功能内置在拦截器和消息侦听器容器中，因此不再提供此建议。

从* version 1.3 *开始，提供了一个构建器 API，以帮助使用 Java(或在@Configuration类中)组装这些拦截器，例如：

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

这种方式只能配置一部分重试功能。更多高级功能需要将RetryTemplate配置为 Spring bean。有关可用策略及其配置的完整信息，请参见Spring 重试 Javadocs。

消息侦听器和异步情况

如果MessageListener因业务异常而失败，则该异常由消息侦听器容器处理，然后返回到侦听另一条消息。如果故障是由断开的连接(不是业务异常)引起的，则必须取消并重新启动为侦听器收集消息的使用者。 SimpleMessageListenerContainer无缝处理此问题，并留下一条日志说正在重新启动侦听器。实际上，它会无休止地循环尝试重新启动使用者，并且只有当使用者表现得很差时，它才会放弃。副作用是，如果在容器启动时代理关闭，则它将 continue 尝试直到可以构建连接为止。

与协议错误和断开连接相反，业务异常处理可能需要更多考虑和一些自定义配置，尤其是在使用事务和/或容器确认的情况下。在 2.8.x 之前，RabbitMQ 没有死信行为的定义，因此默认情况下，由于业务异常而被拒绝或回滚的消息可以无限地重新发送。为了限制 Client 端的重传次数，一种选择是侦听器的建议链中的StatefulRetryOperationsInterceptor。拦截器可以具有实现自定义死信动作的恢复回调：适用于您的特定环境的任何东西。

另一种选择是将容器的defaultRequeueRejected属性设置为 false。这将导致所有失败的消息被丢弃。当使用 RabbitMQ 2.8.x 或更高版本时，这也有助于将消息传递到 Dead Letter Exchange。

或者，您可以抛出AmqpRejectAndDontRequeueException;不管defaultRequeueRejected属性的设置如何，都可以防止消息重新排队。

从 2.1 版开始，引入了ImmediateRequeueAmqpException来执行完全相反的逻辑：无论defaultRequeueRejected属性的设置如何，都会重新排队该消息。

通常，将两种技术结合使用。在建议链中使用StatefulRetryOperationsInterceptor，并使用MessageRecoverer抛出AmqpRejectAndDontRequeueException。所有重试都用尽后，将调用MessageRecover。 RejectAndDontRequeueRecoverer正是这样做的。默认值MessageRecoverer仅消耗错误消息并发出 WARN 消息。

从* version 1.3 *版本开始，提供了一个新的RepublishMessageRecoverer，以允许在重试用尽后发布失败的消息：

当恢复者使用最终异常时，将对消息进行 ACK 确认，并且不会将其发送到 Dead Letter Exchange(如果有的话)。

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

RepublishMessageRecoverer在消息头中发布带有其他信息的消息，例如异常消息，堆栈跟踪，原始交换和路由密钥。可以通过创建子类并覆盖additionalHeaders()来添加其他标题。 deliveryMode(或任何其他属性)也可以在additionalHeaders()中更改：

RepublishMessageRecoverer recoverer = new RepublishMessageRecoverer(amqpTemplate, "error") {

    protected Map<? extends String, ? extends Object> additionalHeaders(Message message, Throwable cause) {
        message.getMessageProperties()
            .setDeliveryMode(message.getMessageProperties().getReceivedDeliveryMode());
        return null;
    }

};

从 2.1 版开始，添加ImmediateRequeueMessageRecoverer引发ImmediateRequeueAmqpException，该ImmediateRequeueAmqpException通知侦听器容器重新排队当前失败的消息。

重试的异常分类

Spring Retry 在确定哪些异常可以调用重试方面具有很大的灵 Active。默认配置将重试所有异常。鉴于用户异常将包含在ListenerExecutionFailedException中，因此我们需要确保分类检查异常原因。默认分类器仅查看顶级异常。

从 Spring Retry 1.0.3 开始，BinaryExceptionClassifier具有属性traverseCauses(默认false)。当true时，它将遍历异常原因，直到找到匹配项或没有原因为止。

要使用此分类器进行重试，请使用使用最大尝试次数的构造函数创建的SimpleRetryPolicy，Exception的Map和布尔值(traverseCauses)，然后将此策略注入RetryTemplate。

3.1.22 Debugging

Spring AMQP 提供了广泛的日志记录，尤其是在DEBUG级别。

如果希望监视应用程序和代理之间的 AMQP 协议，可以使用 WireShark 之类的工具，该工具具有用于解码协议的插件。另外，RabbitMQ JavaClient 端带有一个非常有用的类Tracer。默认情况下，以main身份运行时，它将侦听端口 5673 并连接到 localhost 上的端口 5672.只需运行它，然后更改连接工厂配置以连接到 localhost 上的端口 5673.它在控制台上显示解码的协议。有关更多信息，请参考Tracer javadocs。

3.2 记录子系统 AMQP Appender

该框架为几个流行的日志子系统提供日志附加程序：

• logback(从 Spring AMQP *版本 1.4 *开始)

• log4j2(从 Spring AMQP *版本 1.6 *开始)

附加程序使用日志子系统的常规机制配置，可用属性在以下各节中指定。

3.2.1 通用属性

以下属性可用于所有追加程序：

表 3.4. 常用的附加属性

Property	Default	Description
exchangeName	logs	要将日志事件发布到的交换的名称。
exchangeType	topic	将日志事件发布到的交换类型-仅在附加程序声明交换时才需要。参见declareExchange。
routingKeyPattern	%c.%p	记录子系统模式格式，用于生成路由密钥。
applicationId	``	应用程序 ID-如果模式包含%X{applicationId}，则添加到路由键。
senderPoolSize	2	用于发布日志事件的线程数。
maxSenderRetries	30	如果代理不可用或存在其他错误，请重试发送消息多少次。重试延迟如下：N ^ log(N)，其中N是重试编号。
addresses	``	代理地址的逗号分隔列表：host:port[,host:port]*-覆盖host和port。
host	localhost	要连接的 RabbitMQ 主机。
port	5672	要连接的 RabbitMQ 端口。
virtualHost	/	要连接的 RabbitMQ 虚拟主机。
username	guest	RabbitMQ 用户连接为。
password	guest	该用户的 RabbitMQ 密码。
useSsl	false	对 RabbitMQ 连接使用 SSL。见名为“ RabbitConnectionFactoryBean 和配置 SSL”的部分
sslAlgorithm	null	要使用的 SSL 算法。
sslProperties
位置	null	SSL 属性文件的位置。
keyStore	null	密钥库的位置。
keyStorePassphrase	null	密钥库的密码。
keyStoreType	JKS	密钥库类型。
trustStore	null	信任库的位置。
trustStorePassphrase	null	信任库的密码。
trustStoreType	JKS	信任库类型。
contentType	text/plain	日志消息的content-type属性。
contentEncoding	``	日志消息的content-encoding属性。
declareExchange	false	此附加程序启动时是否声明已配置的交换。另请参见durable和autoDelete。
durable	true	当declareExchange为true时，持久标记设置为此值。
autoDelete	false	当declareExchange为true时，自动删除标志设置为此值。
charset	null	在将 String 转换为 byte []时使用的字符集，默认为 null(使用系统默认字符集)。如果当前平台不支持该字符集，那么我们将转而使用系统字符集。
deliveryMode	PERSISTENT	PERSISTENT 或 NON_PERSISTENT，以确定 RabbitMQ 是否应保留消息。
generateId	false	用于确定messageId属性是否设置为唯一值。
clientConnection 属性	null	用于自定义 Client 端属性到 RabbitMQ 连接的key:value对的逗号分隔列表。

3.2.2 Log4j2 Appender

示例 log4j2.xml 代码片段.

<Appenders>
    ...
    <RabbitMQ name="rabbitmq"
        addresses="foo:5672,bar:5672" user="guest" password="guest" virtualHost="/"
        exchange="log4j2" exchangeType="topic" declareExchange="true" durable="true" autoDelete="false"
        applicationId="myAppId" routingKeyPattern="%X{applicationId}.%c.%p"
        contentType="text/plain" contentEncoding="UTF-8" generateId="true" deliveryMode="NON_PERSISTENT"
        charset="UTF-8"
        senderPoolSize="3" maxSenderRetries="5">
    </RabbitMQ>
</Appenders>

3.2.3 Logback Appender

示例 logback.xml 片段.

<appender name="AMQP" class="org.springframework.amqp.rabbit.logback.AmqpAppender">
    <layout>
        <pattern><![CDATA[ %d %p %t [%c] - <%m>%n ]]></pattern>
    </layout>
    <addresses>foo:5672,bar:5672</addresses>
    <abbreviation>36</abbreviation>
    <includeCallerData>false</includeCallerData>
    <applicationId>myApplication</applicationId>
    <routingKeyPattern>%property{applicationId}.%c.%p</routingKeyPattern>
    <generateId>true</generateId>
    <charset>UTF-8</charset>
    <durable>false</durable>
    <deliveryMode>NON_PERSISTENT</deliveryMode>
    <declareExchange>true</declareExchange>
</appender>

从* version 1.7.1 *开始，Logback AmqpAppender提供了includeCallerData选项，默认情况下为false。提取呼叫者数据可能会非常昂贵，因为日志事件必须创建一个 throwable 并对其进行检查以确定呼叫位置。因此，默认情况下，将事件添加到事件队列时，不会提取与事件关联的呼叫者数据。您可以通过将includeCallerData属性设置为true来配置附加程序以包括调用方数据。

从* version 2.0.0 *开始，Logback AmqpAppender通过encoder选项支持Logback encoders。 encoder和layout选项是互斥的。

3.2.4 自定义消息

默认情况下，AMQP 附加程序将填充以下消息属性：* deliveryMode * contentType * contentEncoding(如果已配置)* messageId(如果已配置generateId 日志事件的时间戳) appId(如果已配置 applicationId)

此外，它们还会填充标题：日志事件的类别名称日志事件的级别线程发生日志事件的线程的名称位置日志事件调用的堆栈跟踪*所有 MDC 属性的副本

每个追加程序都可以被子类化，从而允许您在发布之前修改消息。

自定义日志消息.

public class MyEnhancedAppender extends AmqpAppender {

    @Override
    public Message postProcessMessageBeforeSend(Message message, Event event) {
        message.getMessageProperties().setHeader("foo", "bar");
        return message;
    }

}

3.2.5 自定义 Client 端属性

简单字符串属性

每个追加程序都支持将 Client 端属性添加到 RabbitMQ 连接。

logback.

<appender name="AMQP" ...>
    ...
    <clientConnectionProperties>foo:bar,baz:qux</clientConnectionProperties>
    ...
</appender>

log4j2.

<Appenders>
    ...
    <RabbitMQ name="rabbitmq"
        ...
        clientConnectionProperties="foo:bar,baz:qux"
        ...
    </RabbitMQ>
</Appenders>

这些属性是用逗号分隔的key:value对列表。键和值不能包含逗号或冒号。

查看连接时，这些属性将显示在 RabbitMQManagement 员 UI 上。

高级登录技术

可以将 Logback 附加程序子类化，从而允许您在构建连接之前修改 Client 端连接属性：

自定义 Client 端连接属性.

public class MyEnhancedAppender extends AmqpAppender {

    private String foo;

    @Override
    protected void updateConnectionClientProperties(Map<String, Object> clientProperties) {
        clientProperties.put("foo", this.foo);
    }

    public void setFoo(String foo) {
        this.foo = foo;
    }

}

然后将<foo>bar</foo>添加到 logback.xml。

当然，对于像此示例这样的简单 String 属性，可以使用先前的技术；子类允许更丰富的属性(例如添加Map或数字属性)。

对于 log4j2，由于 log4j2 使用静态工厂方法的方式，因此不支持子类。

3.2.6 提供自定义队列实施

AmqpAppenders使用BlockingQueue将日志记录事件异步发布到 RabbitMQ。默认情况下，使用LinkedBlockingQueue。但是，可以提供任何类型的自定义BlockingQueue实现。

logback.

public class MyEnhancedAppender extends AmqpAppender {

    @Override
    protected BlockingQueue<Event> createEventQueue() {
        return new ArrayBlockingQueue();
    }

}

Log4j2 附加程序支持BlockingQueueFactory的用法。

log4j2.

<Appenders>
    ...
    <RabbitMQ name="rabbitmq"
              bufferSize="10" ... >
        <ArrayBlockingQueue/>
    </RabbitMQ>
</Appenders>

3.3 示例应用程序

3.3.1 Introduction

SpringAMQPsamples项目包括两个示例应用程序。第一个是一个简单的“ Hello World”示例，该示例演示了同步和异步消息接收。它为了解基本组件提供了一个极好的起点。第二个示例基于一个股票 Transaction 用例，以演示在实际应用中常见的交互类型。在本章中，我们将快速介绍每个示例，以便您可以专注于最重要的组件。这些示例都是基于 Maven 的，因此您应该能够将它们直接导入任何支持 Maven 的 IDE(例如SpringSource 工具套件)。

3.3.2 Hello World

Introduction

Hello World 示例演示了同步和异步消息接收。您可以将spring-rabbit-helloworld示例导入到 IDE 中，然后按照下面的讨论进行操作。

Synchronous Example

在src/main/java目录中，导航到org.springframework.amqp.helloworld软件包。打开HelloWorldConfiguration类，注意它在类级别包含@ConfigurationComments，在方法级别包含一些@BeanComments。这是 Spring 基于 Java 的配置的示例。您可以在 http://docs.spring.io/spring/docs/current/spring-framework-reference/html/beans.html#beans-java [此处]了解更多信息。

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

该配置还包含一个实例RabbitAdmin，该实例默认情况下查找类型为 Exchange，Queue 或 Binding 的任何 bean，然后在代理上声明它们。实际上，在 HelloWorldConfiguration 中生成的“ helloWorldQueue” bean 只是一个示例，因为它是 Queue 的一个实例。

@Bean
public Queue helloWorldQueue() {
    return new Queue(this.helloWorldQueueName);
}

回顾“ rabbitTemplate” bean 配置，您将看到它已将 helloWorldQueue 的名称设置为其“ queue”属性(用于接收消息)和其“ routingKey”属性(用于发送消息)。

现在，我们已经探索了配置，让我们看一下实际使用这些组件的代码。首先，从同一包中打开 Producer 类。它包含创建 Spring ApplicationContext的main()方法。

public static void main(String[] args) {
    ApplicationContext context =
        new AnnotationConfigApplicationContext(RabbitConfiguration.class);
    AmqpTemplate amqpTemplate = context.getBean(AmqpTemplate.class);
    amqpTemplate.convertAndSend("Hello World");
    System.out.println("Sent: Hello World");
}

如您在上面的示例中看到的那样，AmqpTemplate bean 被检索并用于发送Message。由于 Client 端代码应尽可能依赖接口，因此类型为AmqpTemplate而不是RabbitTemplate。即使在HelloWorldConfiguration中创建的 bean 是RabbitTemplate的实例，依赖于接口也意味着此代码更具移植性(可以独立于代码而更改配置)。由于调用convertAndSend()方法，因此模板将委派给其MessageConverter实例。在这种情况下，它使用默认的SimpleMessageConverter，但是可以为HelloWorldConfiguration中定义的“ rabbitTemplate” bean 提供不同的实现。

现在打开Consumer类。实际上，它共享相同的配置 Base Class，这意味着它将共享“ rabbitTemplate” bean。这就是为什么我们同时使用“ routingKey”(用于发送)和“ queue”(用于接收)来配置该模板的原因。如您在第 3.1.4 节“ AmqpTemplate”中所见，您可以改为将* routingKey 参数传递给 send 方法，并将 queue *参数传递给 receive 方法。 Consumer代码基本上是生产者的镜像，调用receiveAndConvert()而不是convertAndSend()。

public static void main(String[] args) {
    ApplicationContext context =
        new AnnotationConfigApplicationContext(RabbitConfiguration.class);
    AmqpTemplate amqpTemplate = context.getBean(AmqpTemplate.class);
    System.out.println("Received: " + amqpTemplate.receiveAndConvert());
}

如果您运行Producer，然后运行Consumer，您应该在控制台输出中看到消息“ Received：Hello World”。

Asynchronous Example

现在，我们已经遍历了同步 Hello World 示例，是时候 continue 使用稍微更高级但功能更强大的选项了。经过一些修改，Hello World 示例可以提供一个异步接收的示例，也就是 消息驱动的 POJO 。实际上，有一个子程序包提供了确切的信息：org.springframework.amqp.samples.helloworld.async。

再一次，我们将从发送方开始。打开 ProducerConfiguration 类，注意它创建了一个“ connectionFactory”和“ rabbitTemplate” bean。这次，由于配置专用于消息发送端，因此我们甚至不需要任何队列定义，并且 RabbitTemplate 仅设置了* routingKey *属性。回想一下，邮件是发送到 Exchange 的，而不是直接发送到队列的。 AMQP 默认交换是没有名称的直接交换。所有队列都以其名称作为路由键绑定到该默认 Exchange。这就是为什么我们只需要在此处提供路由密钥的原因。

public RabbitTemplate rabbitTemplate() {
    RabbitTemplate template = new RabbitTemplate(connectionFactory());
    template.setRoutingKey(this.helloWorldQueueName);
    return template;
}

由于此示例将演示异步消息接收，因此产生方将设计为连续发送消息(如果它是像同步版本一样的每次执行消息模型，那么实际上它不是消息-驱动的 Consumer)。负责连续发送消息的组件被定义为ProducerConfiguration中的内部类。它配置为每 3 秒执行一次。

static class ScheduledProducer {

    @Autowired
    private volatile RabbitTemplate rabbitTemplate;

    private final AtomicInteger counter = new AtomicInteger();

    @Scheduled(fixedRate = 3000)
    public void sendMessage() {
        rabbitTemplate.convertAndSend("Hello World " + counter.incrementAndGet());
    }
}

您不需要了解所有细节，因为 true 的重点应该放在接收方(我们将在稍后进行介绍)。但是，如果您还不熟悉 Spring 任务计划支持，则可以了解更多here。简短的故事是 ProducerConfiguration 中的“ postProcessor” bean 正在向调度程序注册任务。

现在，让我们转到接收方。为了强调消息驱动的 POJO 行为，将从响应消息的组件开始。该类称为HelloWorldHandler。

public class HelloWorldHandler {

    public void handleMessage(String text) {
        System.out.println("Received: " + text);
    }

}

显然，这是一个 POJO。它不扩展任何 Base Class，不实现任何接口，甚至不包含任何导入。 Spring AMQP MessageListenerAdapter正在将其“适应”到MessageListener接口。然后可以在SimpleMessageListenerContainer上配置该适配器。对于此示例，该容器是在ConsumerConfiguration类中创建的。您可以在其中看到包装在适配器中的 POJO。

@Bean
public SimpleMessageListenerContainer listenerContainer() {
    SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
    container.setConnectionFactory(connectionFactory());
    container.setQueueName(this.helloWorldQueueName);
    container.setMessageListener(new MessageListenerAdapter(new HelloWorldHandler()));
    return container;
}

SimpleMessageListenerContainer是 Spring 生命周期组件，默认情况下会自动启动。如果查看 Consumer 类，将看到它的main()方法只包含用于创建ApplicationContext的单行引导程序。生产者的main()方法也是单行引导程序，因为使用@ScheduledComments 其方法的组件也将自动开始执行。您可以以任何 Sequences 启动Producer和Consumer，并且应该看到每 3 秒发送和接收一条消息。

3.3.3 股票 Transaction

与“ Hello World”示例相比，“股票 Transaction”示例展示了更高级的消息传递场景。但是，配置非常相似-涉及更多。由于我们已经详细介绍了 Hello World 配置，因此在此我们将重点介绍使此示例与众不同的地方。有一个将市场数据(股票报价)推送到主题 Transaction 所的服务器。然后，Client 可以通过将 Queue 与路由模式(例如app.stock.quotes.nasdaq.*)绑定来订阅市场数据供稿。该演示的另一个主要 Feature 是由 Client 端发起并由服务器处理的请求-答复“股票 Transaction”交互。这涉及由 Client 在订单请求消息本身内发送的私有“ replyTo”队列。

服务器的核心配置位于org.springframework.amqp.rabbit.stocks.config.server包内的RabbitServerConfiguration类中。它扩展了AbstractStockAppRabbitConfiguration。在那里定义了服务器和 Client 端的公共资源，包括市场数据 Topic Exchange(名称为* app.stock.marketdata )和服务器为股票 Transaction 而公开的队列(名称为)。 app.stock.request *)。在该通用配置文件中，您还将看到在RabbitTemplate上配置了Jackson2JsonMessageConverter。

服务器特定的配置由两部分组成。首先，它在RabbitTemplate上配置市场数据交换，因此不需要在每次发送Message的呼叫中都提供该交换名称。它在基本配置类中定义的抽象回调方法中执行此操作。

public void configureRabbitTemplate(RabbitTemplate rabbitTemplate) {
    rabbitTemplate.setExchange(MARKET_DATA_EXCHANGE_NAME);
}

其次，声明库存请求队列。在这种情况下，它不需要任何显式绑定，因为它将绑定到默认的无名称交换，并以其自己的名称作为路由键。如前所述，AMQP 规范定义了该行为。

@Bean
public Queue stockRequestQueue() {
    return new Queue(STOCK_REQUEST_QUEUE_NAME);
}

既然您已经看到了服务器的 AMQP 资源的配置，请导航到src/test/java目录下的org.springframework.amqp.rabbit.stocks包。在那里，您将看到提供main()方法的实际 Server 类。它基于server-bootstrap.xml配置文件创建ApplicationContext。在这里，您将看到发布虚拟市场数据的计划任务。该配置依赖于 Spring 的“任务”名称空间支持。引导程序配置文件还会导入其他一些文件。最有趣的是server-messaging.xml，它直接位于src/main/resources之下。在这里，您将看到负责处理股票 Transaction 请求的“ messageListenerContainer” bean。最后，看看在“ server-handlers.xml”中定义的“ serverHandler” bean(也在* src/main/resources *中)。该 bean 是ServerHandler类的实例，并且是消息驱动的 POJO 的一个很好的示例，该 POJO 也能够发送答复消息。请注意，它本身并不与框架或任何 AMQP 概念耦合。它只接受TradeRequest并返回TradeResponse。

public TradeResponse handleMessage(TradeRequest tradeRequest) { ...
}

现在，我们已经看到了服务器的最重要的配置和代码，下面我们来看 Client 端。最好的起点可能是org.springframework.amqp.rabbit.stocks.config.client软件包中的RabbitClientConfiguration。请注意，它声明了两个队列而没有提供显式名称。

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

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

这些是专用队列，唯一名称将自动生成。Client 端使用第一个生成的队列绑定到服务器公开的市场数据交换。回想一下，在 AMQP 中，Consumer 与队列交互，而生产者与 Exchange 交互。队列与 Exchange 的“绑定”是指示代理将消息从给定 Exchange 传递或路由到队列的内容。由于市场数据交换是主题交换，因此可以使用路由模式来表示绑定。 RabbitClientConfiguration使用Binding对象声明了该对象，并且该对象是使用BindingBuilderFluent 的 API 生成的。

@Value("${stocks.quote.pattern}")
private String marketDataRoutingKey;

@Bean
public Binding marketDataBinding() {
    return BindingBuilder.bind(
        marketDataQueue()).to(marketDataExchange()).with(marketDataRoutingKey);
}

请注意，实际值已在属性文件(src/main/resources 下的“ client.properties”)中外部化，并且我们正在使用 Spring 的@ValueComments 注入该值。这通常是一个好主意，因为否则该值将被硬编码在一个类中，并且无需重新编译就无法修改。在这种情况下，可以更轻松地运行 Client 端的多个版本，同时更改用于绑定的路由模式。让我们现在尝试。

首先运行org.springframework.amqp.rabbit.stocks.Server，然后运行org.springframework.amqp.rabbit.stocks.Client。您应该看到NASDAQ种股票的虚拟报价，因为与 client.properties 中的* stocks.quote.pattern 关键字相关联的当前值为 app.stock.quotes.nasdaq。 **** 现在，在保持现有服务器和 Client 端运行的同时，将该属性值更改为* app.stock.quotes.nyse.**并启动另一个 Client 实例。您应该看到第一个 Client 仍在接收纳斯达克报价，而第二个 Client 仍在接收纽约证券 Transaction 所报价。相反，您可以更改模式以获取所有股票甚至单个股票代码。

我们将探讨的最后一个功能是从 Client 的角度进行请求-答复交互。回想一下，我们已经看到了ServerHandler正在接受TradeRequest对象并返回TradeResponse对象。 Client侧的相应代码是org.springframework.amqp.rabbit.stocks.gateway软件包中的RabbitStockServiceGateway。它委派给RabbitTemplate以便发送消息。

public void send(TradeRequest tradeRequest) {
    getRabbitTemplate().convertAndSend(tradeRequest, new MessagePostProcessor() {
        public Message postProcessMessage(Message message) throws AmqpException {
            message.getMessageProperties().setReplyTo(new Address(defaultReplyToQueue));
            try {
                message.getMessageProperties().setCorrelationId(
                    UUID.randomUUID().toString().getBytes("UTF-8"));
            }
            catch (UnsupportedEncodingException e) {
                throw new AmqpException(e);
            }
            return message;
        }
    });
}

请注意，在发送消息之前，它会设置“ replyTo”地址。它提供了由上面显示的“ traderJoeQueue” bean 定义生成的队列。这是StockServiceGateway类本身的@Bean定义。

@Bean
public StockServiceGateway stockServiceGateway() {
    RabbitStockServiceGateway gateway = new RabbitStockServiceGateway();
    gateway.setRabbitTemplate(rabbitTemplate());
    gateway.setDefaultReplyToQueue(traderJoeQueue());
    return gateway;
}

如果不再运行服务器和 Client 端，请立即启动它们。尝试发送* 100 TCKR *格式的请求。经过短暂的人为延迟，模拟了请求的“处理”，您应该看到一条确认消息出现在 Client 端上。

3.3.4 从非 Spring 应用程序接收 JSON

Spring 应用程序在发送 JSON 时，请将__TypeId__Headers 设置为标准类名，以帮助接收应用程序将 JSON 转换回 Java 对象。

spring-rabbit-json示例探讨了几种从非 Spring 应用程序转换 JSON 的技术。

另请参见名为“ Jackson2JsonMessageConverter”的部分和DefaultClassMapper 的 Javadoc。

3.4 测试支持

3.4.1 Introduction

为异步应用程序编写集成必须比测试更简单的应用程序复杂。当诸如@RabbitListenerComments 之类的抽象引入图片时，这变得更加复杂。问题是如何验证发送消息后，侦听器是否按预期收到了消息。

框架本身具有许多单元测试和集成测试。一些使用模拟，其他使用与实时 RabbitMQ 代理的集成测试。您可以咨询这些测试以获取有关测试方案的一些想法。

Spring AMQP *版本 1.6 *引入了spring-rabbit-test jar，它提供了对测试其中一些更复杂场景的支持。预计该项目将随着时间的推移而扩展，但是我们需要社区的反馈，以便为测试所需的功能提出建议。请使用JIRA或GitHub Issues提供此类反馈。

3.4.2 Mockito 答案\ <?>实现

当前有两种Answer<?>实现可帮助进行测试：

第一个LatchCountDownAndCallRealMethodAnswer提供了Answer<Void>，该Answer<Void>返回null并递减锁存器。

LatchCountDownAndCallRealMethodAnswer answer = new LatchCountDownAndCallRealMethodAnswer(2);
doAnswer(answer)
    .when(listener).foo(anyString(), anyString());

...

assertTrue(answer.getLatch().await(10, TimeUnit.SECONDS));

第二个参数LambdaAnswer<T>提供了一种机制，可以选择调用 real 方法，并提供了基于InvocationOnMock和结果(如果有的话)返回自定义结果的机会。

public class Foo {

    public String foo(String foo) {
        return foo.toUpperCase();
    }

}

Foo foo = spy(new Foo());

doAnswer(new LambdaAnswer<String>(true, (i, r) -> r + r))
    .when(foo).foo(anyString());
assertEquals("FOOFOO", foo.foo("foo"));

doAnswer(new LambdaAnswer<String>(true, (i, r) -> r + i.getArguments()[0]))
    .when(foo).foo(anyString());
assertEquals("FOOfoo", foo.foo("foo"));

doAnswer(new LambdaAnswer<String>(false, (i, r) ->
    "" + i.getArguments()[0] + i.getArguments()[0])).when(foo).foo(anyString());
assertEquals("foofoo", foo.foo("foo"));

使用 Java 7 或更早版本时：

doAnswer(new LambdaAnswer<String>(true, new ValueToReturn<String>() {
    @Override
    public String apply(InvocationOnMock i, String r) {
        return r + r;
    }
})).when(foo).foo(anyString());

3.4.3 @RabbitListenerTest 和 RabbitListenerTestHarness

用@RabbitListenerTestComments 您的@Configuration类之一将使框架用子类RabbitListenerTestHarness替换标准RabbitListenerAnnotationBeanPostProcessor(它还将通过@EnableRabbit启用@RabbitListener检测)。

RabbitListenerTestHarness通过两种方式增强了侦听器-将其包装在Mockito Spy中，从而启用了正常的Mockito存根和验证操作。它还可以向侦听器添加Advice，以允许访问所引发的参数，结果和/或异常。您可以通过@RabbitListenerTest上的属性来控制启用其中的哪一项(或全部启用)。提供后者用于访问有关调用的低级数据-它还支持阻塞测试线程，直到调用异步侦听器为止。

让我们看一些例子。

Using spy:

@Configuration
@RabbitListenerTest
public class Config {

    @Bean
    public Listener listener() {
        return new Listener();
    }

    ...

}

public class Listener {

    @RabbitListener(id="foo", queues="#{queue1.name}")
    public String foo(String foo) {
        return foo.toUpperCase();
    }

    @RabbitListener(id="bar", queues="#{queue2.name}")
    public void foo(@Payload String foo, @Header("amqp_receivedRoutingKey") String rk) {
        ...
    }

}

public class MyTests {

    @Autowired
    private RabbitListenerTestHarness harness; (1)

    @Test
    public void testTwoWay() throws Exception {
        assertEquals("FOO", this.rabbitTemplate.convertSendAndReceive(this.queue1.getName(), "foo"));

        Listener listener = this.harness.getSpy("foo"); (2)
        assertNotNull(listener);
        verify(listener).foo("foo");
    }

    @Test
    public void testOneWay() throws Exception {
        Listener listener = this.harness.getSpy("bar");
        assertNotNull(listener);

        LatchCountDownAndCallRealMethodAnswer answer = new LatchCountDownAndCallRealMethodAnswer(2); (3)
        doAnswer(answer).when(listener).foo(anyString(), anyString()); (4)

        this.rabbitTemplate.convertAndSend(this.queue2.getName(), "bar");
        this.rabbitTemplate.convertAndSend(this.queue2.getName(), "baz");

        assertTrue(answer.getLatch().await(10, TimeUnit.SECONDS));
        verify(listener).foo("bar", this.queue2.getName());
        verify(listener).foo("baz", this.queue2.getName());
    }

}

• (1) 将线束注入测试用例中，以便我们可以访问 Spy。
• (2) 获取对 Spy 的引用，以便我们可以验证它是否按预期被调用。由于这是一个发送和接收操作，因此无需挂起测试线程，因为它已在RabbitTemplate中挂起，await 答复。
• (3) 在这种情况下，我们仅使用发送操作，因此我们需要一个闩锁来 await 对容器线程上的侦听器的异步调用。我们使用Answer<?>实现之一来帮助您。
• (4) 配置 Spy 以调用Answer。

使用捕获建议：

@Configuration
@ComponentScan
@RabbitListenerTest(spy = false, capture = true)
public class Config {

}

@Service
public class Listener {

    private boolean failed;

    @RabbitListener(id="foo", queues="#{queue1.name}")
    public String foo(String foo) {
        return foo.toUpperCase();
    }

    @RabbitListener(id="bar", queues="#{queue2.name}")
    public void foo(@Payload String foo, @Header("amqp_receivedRoutingKey") String rk) {
        if (!failed && foo.equals("ex")) {
            failed = true;
            throw new RuntimeException(foo);
        }
        failed = false;
    }

}

public class MyTests {

    @Autowired
    private RabbitListenerTestHarness harness; (1)

    @Test
    public void testTwoWay() throws Exception {
        assertEquals("FOO", this.rabbitTemplate.convertSendAndReceive(this.queue1.getName(), "foo"));

        InvocationData invocationData =
            this.harness.getNextInvocationDataFor("foo", 0, TimeUnit.SECONDS); (2)
        assertThat(invocationData.getArguments()[0], equalTo("foo"));     (3)
        assertThat((String) invocationData.getResult(), equalTo("FOO"));
    }

    @Test
    public void testOneWay() throws Exception {
        this.rabbitTemplate.convertAndSend(this.queue2.getName(), "bar");
        this.rabbitTemplate.convertAndSend(this.queue2.getName(), "baz");
        this.rabbitTemplate.convertAndSend(this.queue2.getName(), "ex");

        InvocationData invocationData =
            this.harness.getNextInvocationDataFor("bar", 10, TimeUnit.SECONDS); (4)
        Object[] args = invocationData.getArguments();
        assertThat((String) args[0], equalTo("bar"));
        assertThat((String) args[1], equalTo(queue2.getName()));

        invocationData = this.harness.getNextInvocationDataFor("bar", 10, TimeUnit.SECONDS);
        args = invocationData.getArguments();
        assertThat((String) args[0], equalTo("baz"));

        invocationData = this.harness.getNextInvocationDataFor("bar", 10, TimeUnit.SECONDS);
        args = invocationData.getArguments();
        assertThat((String) args[0], equalTo("ex"));
        assertEquals("ex", invocationData.getThrowable().getMessage()); (5)
    }

}

• (1) 将线束注入测试用例中，以便我们可以访问 Spy。
• (2) 使用harness.getNextInvocationDataFor()来检索调用数据-在这种情况下，由于它是一个请求/答复方案，因此无需 await 任何时间，因为测试线程已挂在RabbitTemplate中 await 结果。
• (3) 然后，我们可以验证参数和结果是否符合预期。
• (4) 这次我们需要一些时间来 await 数据，因为它是对容器线程的异步操作，因此我们需要挂起测试线程。
• (5) 侦听器引发异常时，在调用数据的throwable属性中可用。

3.4.4 TestRabbitTemplate

提供TestRabbitTemplate可以执行一些基本的集成测试，而无需代理。在测试用例中将其添加为@Bean时，它将发现上下文中的所有侦听器容器，这些容器都声明为@Bean，<bean/>`s or using the `@RabbitListenerComments。当前仅支持按队列名称进行路由。该模板从容器中提取消息侦听器，然后直接在测试线程上调用它。返回回复的侦听器支持请求/回复消息传递(sendAndReceive方法)。

这是一个使用模板的简单测试用例：

@RunWith(SpringRunner.class)
public class TestRabbitTemplateTests {

    @Autowired
    private TestRabbitTemplate template;

    @Autowired
    private Config config;

    @Test
    public void testSimpleSends() {
        this.template.convertAndSend("foo", "hello1");
        assertThat(this.config.fooIn, equalTo("foo:hello1"));
        this.template.convertAndSend("bar", "hello2");
        assertThat(this.config.barIn, equalTo("bar:hello2"));
        assertThat(this.config.smlc1In, equalTo("smlc1:"));
        this.template.convertAndSend("foo", "hello3");
        assertThat(this.config.fooIn, equalTo("foo:hello1"));
        this.template.convertAndSend("bar", "hello4");
        assertThat(this.config.barIn, equalTo("bar:hello2"));
        assertThat(this.config.smlc1In, equalTo("smlc1:hello3hello4"));

        this.template.setBroadcast(true);
        this.template.convertAndSend("foo", "hello5");
        assertThat(this.config.fooIn, equalTo("foo:hello1foo:hello5"));
        this.template.convertAndSend("bar", "hello6");
        assertThat(this.config.barIn, equalTo("bar:hello2bar:hello6"));
        assertThat(this.config.smlc1In, equalTo("smlc1:hello3hello4hello5hello6"));
    }

    @Test
    public void testSendAndReceive() {
        assertThat(this.template.convertSendAndReceive("baz", "hello"), equalTo("baz:hello"));
    }

@Configuration
    @EnableRabbit
    public static class Config {

        public String fooIn = "";

        public String barIn = "";

        public String smlc1In = "smlc1:";

        @Bean
        public TestRabbitTemplate template() throws IOException {
            return new TestRabbitTemplate(connectionFactory());
        }

        @Bean
        public ConnectionFactory connectionFactory() throws IOException {
            ConnectionFactory factory = mock(ConnectionFactory.class);
            Connection connection = mock(Connection.class);
            Channel channel = mock(Channel.class);
            willReturn(connection).given(factory).createConnection();
            willReturn(channel).given(connection).createChannel(anyBoolean());
            given(channel.isOpen()).willReturn(true);
            return factory;
        }

        @Bean
        public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory() throws IOException {
            SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
            factory.setConnectionFactory(connectionFactory());
            return factory;
        }

        @RabbitListener(queues = "foo")
        public void foo(String in) {
            this.fooIn += "foo:" + in;
        }

        @RabbitListener(queues = "bar")
        public void bar(String in) {
            this.barIn += "bar:" + in;
        }

        @RabbitListener(queues = "baz")
        public String baz(String in) {
            return "baz:" + in;
        }

        @Bean
        public SimpleMessageListenerContainer smlc1() throws IOException {
            SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory());
            container.setQueueNames("foo", "bar");
            container.setMessageListener(new MessageListenerAdapter(new Object() {

                @SuppressWarnings("unused")
                public void handleMessage(String in) {
                    smlc1In += in;
                }

            }));
            return container;
        }

    }

}

3.4.5 JUnit4 @Rules

Spring AMQP *版本 1.7 *及更高版本提供了一个额外的 jar spring-rabbit-junit;该 jar 包含几个 Util@Rule，可在运行 JUnit4 测试时使用。有关 JUnit5 测试，请参见第 3.4.6 节“ JUnit5 条件”。

BrokerRunning

BrokerRunning提供了一种机制，当代理未运行时(默认情况下为localhost)，测试可以成功进行。

它还具有 Util 方法来初始化/清空队列，以及删除队列和交换。

Usage:

@ClassRule
public static BrokerRunning brokerRunning = BrokerRunning.isRunningWithEmptyQueues("foo", "bar");

@AfterClass
public static void tearDown() {
    brokerRunning.removeTestQueues("some.other.queue.too") // removes foo, bar as well
}

有几种isRunning...静态方法(例如isBrokerAndManagementRunning())可以验证代理是否启用了 Management 插件。

配置规则

有时候，如果没有代理，您希望测试失败，例如每晚配置项构建。要在运行时禁用规则，请将环境变量RABBITMQ_SERVER_REQUIRED设置为true。

您可以通过几种方式覆盖代理属性，例如主机名：

• Setters

@ClassRule
public static BrokerRunning brokerRunning = BrokerRunning.isRunningWithEmptyQueues("foo", "bar");

static {
    brokerRunning.setHostName("10.0.0.1")
}

@AfterClass
public static void tearDown() {
    brokerRunning.removeTestQueues("some.other.queue.too") // removes foo, bar as well
}

• Environment Variables

提供了以下环境变量：

public static final String BROKER_ADMIN_URI = "RABBITMQ_TEST_ADMIN_URI";
public static final String BROKER_HOSTNAME = "RABBITMQ_TEST_HOSTNAME";
public static final String BROKER_PORT = "RABBITMQ_TEST_PORT";
public static final String BROKER_USER = "RABBITMQ_TEST_USER";
public static final String BROKER_PW = "RABBITMQ_TEST_PASSWORD";
public static final String BROKER_ADMIN_USER = "RABBITMQ_TEST_ADMIN_USER";
public static final String BROKER_ADMIN_PW = "RABBITMQ_TEST_ADMIN_PASSWORD";

这些将覆盖默认设置(对于 amqp 来说是localhost:5672，对于 ManagementREST API 来说是http://localhost:15672/api/)。

更改主机名会影响 amqp 和 ManagementREST API 连接(除非显式设置了 admin uri)。

BrokerRunning还提供了static方法：setEnvironmentVariableOverrides，您可以在其中传递包含这些变量的 Map；它们覆盖系统环境变量。如果您希望对多个测试套件中的测试使用不同的配置，这可能会很有用。重要说明：在调用任何创建规则实例的isRunning()静态方法之前，必须先调用该方法。变量值将应用于此后创建的所有实例。调用clearEnvironmentVariableOverrides()以将规则重置为使用默认值(包括任何实际的环境变量)。

在测试用例中，可以在创建连接工厂时使用这些属性：

@Bean
public ConnectionFactory rabbitConnectionFactory() {
    CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
    connectionFactory.setHost(brokerRunning.getHostName());
    connectionFactory.setPort(brokerRunning.getPort());
    connectionFactory.setUsername(brokerRunning.getUser());
    connectionFactory.setPassword(brokerRunning.getPassword());
    return connectionFactory;
}

LongRunningIntegrationTest

LongRunningIntegrationTest是禁用长时间运行测试的规则；您可能要在开发人员系统上使用此功能，但请确保在例如夜间 CI 构建中禁用该规则。

Usage:

@Rule
public LongRunningIntegrationTest longTests = new LongRunningIntegrationTest();

要在运行时禁用规则，请将环境变量RUN_LONG_INTEGRATION_TESTS设置为true。

3.4.6 JUnit5 条件

• 2.0.2 *版引入了对 JUnit5 的支持。

@RabbitAvailable Annotation

这个类级别的 Comments 类似于第 3.4.5 节“ JUnit4 @Rules”中讨论的BrokerRunning @Rule；它由RabbitAvailableCondition处理。

注解具有两个属性：

• queues-将在每个测试之前声明(和清除)队列，并在所有测试完成后将其删除。

• management-如果您的测试还需要在代理上安装 Management 插件，则设置为true。

它用于检查代理是否可用，如果没有，则跳过测试。如称为“配置规则”的部分中的讨论，环境变量RABBITMQ_SERVER_REQUIRED，如果没有代理，则true将导致测试快速失败。可以使用称为“配置规则”的部分中讨论的环境变量来配置条件。

此外，RabbitAvailableCondition支持参数化测试构造函数和方法的参数解析。支持两种参数类型BrokerRunning(实例)和ConnectionFactory-BrokerRunning的 RabbitMQ 连接工厂。

这是每个示例：

@RabbitAvailable(queues = "rabbitAvailableTests.queue")
public class RabbitAvailableCTORInjectionTests {

    private final ConnectionFactory connectionFactory;

    public RabbitAvailableCTORInjectionTests(BrokerRunning brokerRunning) {
        this.connectionFactory = brokerRunning.getConnectionFactory();
    }

    @Test
    public void test(ConnectionFactory cf) throws Exception {
        assertSame(cf, this.connectionFactory);
        Connection conn = this.connectionFactory.newConnection();
        Channel channel = conn.createChannel();
        DeclareOk declareOk = channel.queueDeclarePassive("rabbitAvailableTests.queue");
        assertEquals(0, declareOk.getConsumerCount());
        channel.close();
        conn.close();
    }

}

此测试在框架本身中，并验证参数注入以及条件是否正确创建了队列。

实际的用户测试可能是：

@RabbitAvailable(queues = "rabbitAvailableTests.queue")
public class RabbitAvailableCTORInjectionTests {

    private final CachingConnectionFactory connectionFactory;

    public RabbitAvailableCTORInjectionTests(BrokerRunning brokerRunning) {
        this.connectionFactory =
            new CachingConnectionFactory(brokerRunning.getConnectionFactory());
    }

    @Test
    public void test() throws Exception {
        RabbitTemplate template = new RabbitTemplate(this.connectionFactory);
        ...
    }
}

在测试类中使用 SpringComments 应用程序上下文时，还可以通过静态方法RabbitAvailableCondition.getBrokerRunning()获得对条件连接工厂的引用。这是框架中的另一个测试，演示了用法：

@RabbitAvailable(queues = {
        RabbitTemplateMPPIntegrationTests.QUEUE,
        RabbitTemplateMPPIntegrationTests.REPLIES })
@SpringJUnitConfig
@DirtiesContext(classMode = ClassMode.AFTER_EACH_TEST_METHOD)
public class RabbitTemplateMPPIntegrationTests {

    public static final String QUEUE = "mpp.tests";

    public static final String REPLIES = "mpp.tests.replies";

    @Autowired
    private RabbitTemplate template;

    @Autowired
    private Config config;

    @Test
    public void test() {

        ...

    }

    @Configuration
    @EnableRabbit
    public static class Config {

        @Bean
        public CachingConnectionFactory cf() {
            return new CachingConnectionFactory(RabbitAvailableCondition
                    .getBrokerRunning()
                    .getConnectionFactory());
        }

        @Bean
        public RabbitTemplate template() {

            ...

        }

        @Bean
        public SimpleRabbitListenerContainerFactory
                            rabbitListenerContainerFactory() {

            ...

        }

        @RabbitListener(queues = QUEUE)
        public byte[] foo(byte[] in) {
            return in;
        }

    }

}

@LongRunning Annotation

与LongRunningIntegrationTest JUnit4 @Rule相似，此 Comments 会导致跳过测试，除非将环境变量(或系统属性)设置为true。

@RabbitAvailable(queues = SimpleMessageListenerContainerLongTests.QUEUE)
@LongRunning
public class SimpleMessageListenerContainerLongTests {

    public static final String QUEUE = "SimpleMessageListenerContainerLongTests.queue";

...

}

默认情况下，变量为RUN_LONG_INTEGRATION_TESTS，但是可以在 Comments 的value属性中指定变量名称。