连接与资源管理

Doubao 中英对照 Connection and Resource Management

上一节描述的 AMQP 模型是通用的、适用于所有实现，但进入资源管理时，细节就是 broker 实现所特有的了。因此本节聚焦于仅存在于我们“spring-rabbit” 模块中的代码，因为目前 RabbitMQ 是唯一支持的实现。

管理到 RabbitMQ broker 连接的核心组件是 ConnectionFactory 接口。ConnectionFactory 实现的职责是提供 org.springframework.amqp.rabbit.connection.Connection 的实例，它是 com.rabbitmq.client.Connection 的包装。

选择连接工厂

有三个连接工厂可供选择：

• PooledChannelConnectionFactory
• ThreadChannelConnectionFactory
• CachingConnectionFactory

前两个是在 2.3 版本中添加的。

对于大多数用例，应使用 CachingConnectionFactory。如果你想确保严格的消息顺序而无需使用作用域操作，可以使用 ThreadChannelConnectionFactory。PooledChannelConnectionFactory 与 CachingConnectionFactory 类似，在于它使用单个连接和一组 channel 池。它的实现更简单，但不支持关联的发布确认。与 channel 缓存默认无界的 CachingConnectionFactory 不同，PooledChannelConnectionFactory 使用有界池，在没有可用 channel 时阻塞。容量考量参见 PooledChannelConnectionFactory。

简单发布确认由所有三个工厂支持。

配置 RabbitTemplate 使用单独的连接时，从 2.3.2 版本开始，你现在可以把发布连接工厂配置为不同类型。默认情况下，发布工厂类型相同，在主工厂上设置的任何属性也会传播到发布工厂。

从 3.1 版本开始，AbstractConnectionFactory 包含 connectionCreatingBackOff 属性，在连接模块中支持退避策略。目前，在 createChannel() 的行为中支持处理达到 channelMax 限制时发生的异常，实现基于尝试次数和间隔的退避策略。

PooledChannelConnectionFactory

此工厂管理单个连接和两个基于 Apache Pool2 的 channel 池。一个池用于事务性 channel，另一个用于非事务性 channel。这些池是使用默认配置的 GenericObjectPool；提供了一个回调来配置这些池；更多信息参见 Apache 文档。

要使用此工厂，类路径上必须有 Apache commons-pool2 jar。

@Bean
PooledChannelConnectionFactory pcf() throws Exception {
    ConnectionFactory rabbitConnectionFactory = new ConnectionFactory();
    rabbitConnectionFactory.setHost("localhost");
    PooledChannelConnectionFactory pcf = new PooledChannelConnectionFactory(rabbitConnectionFactory);
    pcf.setPoolConfigurer((pool, tx) -> {
        if (tx) {
            // configure the transactional pool
        }
        else {
            // configure the non-transactional pool
        }
    });
    return pcf;
}

important

这些池使用 Apache Pool2 默认配置，允许每个池最多八个 channel，池耗尽时无限期阻塞。请根据应用所需的并发调整池大小；否则，线程会阻塞且无异常、无日志消息，看起来就像应用停滞了。使用请求/回复和默认的 DirectReplyToMessageListenerContainer 时，每个在途请求在整个交换期间持有一个 channel，因此池必须至少与并发请求数一样大。关于每次交换不持有 channel 的替代方案，参见 RabbitMQ Direct reply-to。

备注

此工厂不适合消费者。监听器容器在运行期间一直持有其 channel，这会将其永久移出池。消费者请使用 CachingConnectionFactory，或为发布侧配置单独连接；参见使用单独的连接。

ThreadChannelConnectionFactory

此工厂管理单个连接和两个 ThreadLocal，一个用于事务性 channel，另一个用于非事务性 channel。此工厂确保同一线程上的所有操作使用同一个 channel（只要它保持打开）。这便于严格的消息顺序，而无需作用域操作。为避免内存泄漏，如果你的应用使用许多短生命周期线程，必须调用工厂的 closeThreadChannel() 释放 channel 资源。从 2.3.7 版本开始，一个线程可以把其 channel 转移到另一个线程。更多信息参见多线程环境中的严格消息顺序。

CachingConnectionFactory

提供的第三个实现是 CachingConnectionFactory，默认建立一个可被应用共享的单一连接代理。连接之所以可共享，是因为 AMQP 消息传递的“工作单元”实际上是“channel”（在某些方面，这类似于 JMS 中连接与会话的关系）。连接实例提供 createChannel 方法。CachingConnectionFactory 实现支持缓存这些 channel，并根据 channel 是否事务性维护单独的缓存。创建 CachingConnectionFactory 实例时，你可以通过构造函数提供 'hostname'。你还应提供 'username' 和 'password' 属性。要配置 channel 缓存大小（默认 25），可以调用 setChannelCacheSize() 方法。

从 1.3 版本开始，你可以配置 CachingConnectionFactory 既缓存连接也只缓存 channel。在这种情况下，每次调用 createConnection() 都会创建一个新连接（或从缓存取出一个空闲连接）。关闭一个连接会把它归还缓存（如果未达到缓存大小）。在此类连接上创建的 channel 也会被缓存。在某些环境中，使用单独连接可能有用，例如从 HA 集群消费时配合负载均衡连接到不同集群节点等。要缓存连接，把 cacheMode 设置为 CacheMode.CONNECTION。

备注

这不限制连接数。相反，它指定允许多少个空闲的打开连接。

从 1.5.5 版本开始，提供了一个名为 connectionLimit 的新属性。设置此属性时，它限制允许的连接总数。设置后，如果达到限制，会使用 channelCheckoutTimeLimit 等待连接变为空闲。如果超时，会抛出 AmqpTimeoutException。

important

当缓存模式为 CONNECTION 时，不支持队列等的自动声明（参见交换机、队列和绑定的自动声明）。

此外，在撰写本文时，amqp-client 库默认会为每个连接创建固定线程池（默认大小：Runtime.getRuntime().availableProcessors() * 2 个线程）。使用大量连接时，你应考虑在 CachingConnectionFactory 上设置自定义 executor。这样，所有连接可以使用同一个 executor，其线程可共享。executor 的线程池应无界，或根据预期用途适当设置（通常每个连接至少一个线程）。如果每个连接上创建多个 channel，池大小会影响并发，因此可变（或简单缓存）线程池执行器最合适。

重要的是要理解，缓存大小（默认）不是一个上限，而仅仅是可缓存的 channel 数量。缓存大小比如为 10 时，实际上可以使用任意数量的 channel。如果使用超过 10 个 channel 且它们都归还缓存，则 10 个进入缓存，其余被物理关闭。

从 1.6 版本开始，默认 channel 缓存大小从 1 增加到 25。在高吞吐量、多线程环境中，小缓存意味着 channel 以高速率创建和关闭。增大默认缓存大小可以避免这种开销。你应通过 RabbitMQ Admin UI 监控在用的 channel，如果看到大量 channel 被创建和关闭，考虑进一步增大缓存大小。缓存只按需增长（以适应应用的并发需求），因此此更改不影响现有的低吞吐量应用。

从 1.4.2 版本开始，CachingConnectionFactory 有一个名为 channelCheckoutTimeout 的属性。当此属性大于零时，channelCacheSize 成为可在一个连接上创建 channel 数量的上限。如果达到限制，调用线程会阻塞，直到有 channel 可用或此超时到期，此时抛出 AmqpTimeoutException。

注意

框架内使用的 channel（例如 RabbitTemplate）会被可靠地归还缓存。如果你在框架外创建 channel（例如直接访问连接并调用 createChannel()），必须可靠地归还它们（通过关闭），也许在 finally 块中，以避免 channel 耗尽。

下例展示了如何创建新的 connection：

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>

备注

还有一个 SingleConnectionFactory 实现，仅在框架的单元测试代码中可用。它比 CachingConnectionFactory 简单，因为它不缓存 channel，但由于缺乏性能和弹性，它不适合在简单测试之外实际使用。如果你因某种原因需要实现自己的 ConnectionFactory，AbstractConnectionFactory 基类可能是一个不错的起点。

可以使用 rabbit 命名空间快速便捷地创建 ConnectionFactory，如下所示：

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

在大多数情况下，这种方式更可取，因为框架可以为你选择最佳默认值。创建的实例是 CachingConnectionFactory。请记住，channel 的默认缓存大小为 25。如果你希望缓存更多 channel，通过设置 '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"/>

你可以使用命名空间提供 host 和 port 属性，如下所示：

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

或者，如果运行在集群环境中，可以使用 addresses 属性，如下所示：

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

关于 address-shuffle-mode 的信息，参见连接到集群。

下例带一个自定义线程工厂，把线程名前缀设为 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>

AddressResolver

从 2.1.15 版本开始，你现在可以使用 AddressResolver 解析连接地址。这会覆盖 addresses 和 host/port 属性的任何设置。

命名连接

从 1.7 版本开始，提供了 ConnectionNameStrategy，可注入 AbstractionConnectionFactory。生成的名称用于应用特定地标识目标 RabbitMQ 连接。如果 RabbitMQ 服务器支持，连接名会显示在管理 UI 中。此值不必唯一，也不能用作连接标识符——例如在 HTTP API 请求中。此值应是人类可读的，它是 ClientProperties 中 connection_name 键下的一部分。你可以使用简单的 Lambda，如下所示：

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

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

SimplePropertyValueConnectionNameStrategy 的一个实现把连接名设置为某个应用属性。你可以把它声明为 @Bean 并注入连接工厂，如下例所示：

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

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

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

备注

使用 Spring Boot 及其自动配置的连接工厂时，你只需声明 ConnectionNameStrategy @Bean。Spring Boot 自动检测该 bean 并把它装配到工厂中。

阻塞连接与资源约束

连接可能因对应内存告警的 broker 交互而被阻塞。从 2.0 版本开始，可以为 org.springframework.amqp.rabbit.connection.Connection 提供 com.rabbitmq.client.BlockedListener 实例，以接收连接阻塞和解除阻塞事件的通知。此外，AbstractConnectionFactory 通过其内部 BlockedListener 实现分别发出 ConnectionBlockedEvent 和 ConnectionUnblockedEvent。这些让你可以提供应用逻辑，对 broker 上的问题做出适当反应，并（例如）采取一些纠正措施。

important

当应用配置为单个 CachingConnectionFactory（Spring Boot 自动配置默认如此）时，连接被 Broker 阻塞时应用停止工作。而它被 Broker 阻塞时，其任何客户端都停止工作。如果我们在同一应用中有生产者和消费者，当生产者阻塞连接（因为 Broker 上不再有资源）而消费者无法释放它们（因为连接被阻塞）时，最终可能死锁。为缓解此问题，我们建议再单独建一个相同选项的 CachingConnectionFactory 实例——一个给生产者，一个给消费者。对于在消费者线程上执行的事务性生产者，单独的 CachingConnectionFactory 不可行，因为它们应复用与消费者事务关联的 Channel。

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

从 1.7.7 版本开始，提供了 AmqpResourceNotAvailableException，当 SimpleConnection.createChannel() 无法创建 Channel 时（例如因为达到 channelMax 限制且缓存中没有可用 channel）抛出。你可以在 RetryPolicy 中使用此异常，在一些退避之后恢复操作。

配置底层客户端连接工厂

CachingConnectionFactory 使用 Rabbit 客户端 ConnectionFactory 的实例。在 CachingConnectionFactory 上设置等效属性时，许多配置属性会被传递（例如 host、port、userName、password、requestedHeartBeat 和 connectionTimeout）。要设置其他属性（例如 clientProperties），可以定义 Rabbit 工厂的实例，并使用 CachingConnectionFactory 的相应构造函数提供对它的引用。使用命名空间（如前所述）时，需要在 connection-factory 属性中提供对已配置工厂的引用。为方便起见，提供了一个工厂 bean 来帮助在 Spring 应用上下文中配置连接工厂，如下一节所述。

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

备注

4.0.x 客户端默认启用自动恢复。虽然与此功能兼容，但 Spring AMQP 有自己的恢复机制，客户端恢复功能通常不需要。我们建议禁用 amqp-client 自动恢复，以避免在 broker 可用但连接尚未恢复时得到 AutoRecoverConnectionNotCurrentlyOpenException 实例。你可能会注意到此异常，例如在 RabbitTemplate 中配置了 RetryTemplate 时，即使是在集群中故障转移到另一个 broker 时。由于自动恢复连接按定时器恢复，使用 Spring AMQP 的恢复机制可能更快恢复连接。从 1.7.1 版本开始，Spring AMQP 会禁用 amqp-client 自动恢复，除非你显式创建自己的 RabbitMQ 连接工厂并提供给 CachingConnectionFactory。由 RabbitConnectionFactoryBean 创建的 RabbitMQ ConnectionFactory 实例默认也禁用此选项。

RabbitConnectionFactoryBean 与配置 SSL

从 1.4 版本开始，提供了一个便捷的 RabbitConnectionFactoryBean，可通过依赖注入方便地配置底层客户端连接工厂上的 SSL 属性。其他 setter 委托给底层工厂。以前，你必须以编程方式配置 SSL 选项。下例展示了如何配置 RabbitConnectionFactoryBean：

Java

@Bean
RabbitConnectionFactoryBean rabbitConnectionFactory() {
    RabbitConnectionFactoryBean factoryBean = new RabbitConnectionFactoryBean();
    factoryBean.setUseSSL(true);
    factoryBean.setSslPropertiesLocation(new ClassPathResource("secrets/rabbitSSL.properties"));
    return factoryBean;
}

@Bean
CachingConnectionFactory connectionFactory(ConnectionFactory rabbitConnectionFactory) {
    CachingConnectionFactory ccf = new CachingConnectionFactory(rabbitConnectionFactory);
    ccf.setHost("...");
    // ...
    return ccf;
}

XML

<bean id="rabbitConnectionFactory"
        class="org.springframework.amqp.rabbit.connection.RabbitConnectionFactoryBean">
    <property name="useSSL" value="true" />
    <property name="sslPropertiesLocation" value="classpath:secrets/rabbitSSL.properties"/>
</bean>

<rabbit:connection-factory id="connectionFactory"
    connection-factory="rabbitConnectionFactory"
    host="${host}"
    port="${port}"
    virtual-host="${vhost}"
    username="${username}" password="${password}" />

Spring Boot 应用文件（.yaml 或 .properties）

Properties

spring.rabbitmq.host=...
spring.rabbitmq.ssl.keyStoreType=jks
spring.rabbitmq.ssl.trustStoreType=jks
spring.rabbitmq.ssl.keyStore=...
spring.rabbitmq.ssl.trustStore=...
spring.rabbitmq.ssl.trustStorePassword=...
spring.rabbitmq.ssl.keyStorePassword=...
spring.rabbitmq.ssl.enabled=true

YAML

spring:
  rabbitmq:
    host: ...
    ssl:
      keyStoreType: jks
      trustStoreType: jks
      keyStore: ...
      trustStore: ...
      trustStorePassword: ...
      keyStorePassword: ...
      enabled: true

关于配置 SSL 的信息，参见 RabbitMQ 文档。省略 keyStore 和 trustStore 配置即可通过 SSL 连接而不做证书验证。下例展示了如何提供密钥库和信任库配置。

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

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

keyStore 和 truststore 是指向这些库的 Spring Resource。通常此属性文件由操作系统保护，应用有读取权限。

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

important

从 2.0 版本开始，服务器证书默认被验证，因为这样更安全。如果你因某种原因希望跳过此验证，把工厂 bean 的 skipServerCertificateValidation 属性设置为 true。从 2.1 版本开始，RabbitConnectionFactoryBean 现在默认调用 enableHostnameVerification()。要恢复以前行为，把 enableHostnameVerification 属性设置为 false。

important

从 2.2.5 版本开始，工厂 bean 默认始终使用 TLS v1.2；以前它在某些情况下使用 v1.1，在另一些情况下使用 v1.2（取决于其他属性）。如果你因某种原因需要使用 v1.1，设置 sslAlgorithm 属性：setSslAlgorithm("TLSv1.1")。

连接到集群

要连接到集群，在 CachingConnectionFactory 上配置 addresses 属性：

@Bean
public CachingConnectionFactory ccf() {
    CachingConnectionFactory ccf = new CachingConnectionFactory();
    ccf.setAddresses("host1:5672,host2:5672,host3:5672");
    return ccf;
}

从 3.0 版本开始，每当建立新连接时，底层连接工厂会通过选择随机地址尝试连接到某个主机。要恢复以前从第一个到最后一个依次尝试连接的行为，把 addressShuffleMode 属性设置为 AddressShuffleMode.NONE。

从 2.3 版本开始，添加了 INORDER 洗牌模式，即创建连接后把第一个地址移到末尾。如果你希望配合RabbitMQ 分片插件、CacheMode.CONNECTION 和合适的并发从所有节点上的所有分片消费，可以使用此模式。

@Bean
public CachingConnectionFactory ccf() {
    CachingConnectionFactory ccf = new CachingConnectionFactory();
    ccf.setAddresses("host1:5672,host2:5672,host3:5672");
    ccf.setAddressShuffleMode(AddressShuffleMode.INORDER);
    return ccf;
}

路由连接工厂

从 1.3 版本开始，引入了 AbstractRoutingConnectionFactory。此工厂提供一种机制来为若干 ConnectionFactories 配置映射，并在运行时通过某个 lookupKey 确定目标 ConnectionFactory。通常，该实现检查线程绑定的上下文。为方便起见，Spring AMQP 提供了 SimpleRoutingConnectionFactory，它从 SimpleResourceHolder 获取当前线程绑定的 lookupKey。以下示例展示了如何在 XML 和 Java 中配置 SimpleRoutingConnectionFactory：

<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 的 JavaDoc。

从 1.4 版本开始，RabbitTemplate 支持 SpEL sendConnectionFactorySelectorExpression 和 receiveConnectionFactorySelectorExpression 属性，它们在每次 AMQP 协议交互操作（send、sendAndReceive、receive 或 receiveAndReply）时求值，解析为所提供 AbstractRoutingConnectionFactory 的 lookupKey 值。你可以在表达式中使用 bean 引用，例如 @vHostResolver.getVHost(#root)。对于 send 操作，待发送消息是根求值对象。对于 receive 操作，queueName 是根求值对象。

路由算法如下：如果选择器表达式为 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 属性。

此外，从 1.4 版本开始，你可以在监听器容器中配置路由连接工厂。在这种情况下，队列名列表用作查找键。例如，如果用 setQueueNames("thing1", "thing2") 配置容器，查找键为 [thing1,thing]"（注意键中没有空格）。

从 1.6.9 版本开始，你可以在监听器容器上使用 setLookupKeyQualifier 为查找键添加限定符。这样做可以实现例如监听同名但位于不同虚拟主机的队列（在那里你会为每个主机配置一个连接工厂）。

例如，查找键限定符为 thing1、容器监听队列 thing2 时，你可以用 thing1[thing2] 作为查找键注册目标连接工厂。

important

目标（以及默认，如果提供的话）连接工厂在发布确认和退回方面必须具有相同设置。参见发布确认与退回。

从 2.4.4 版本开始，可以禁用此校验。如果你遇到确认和退回之间的值需要不相等的情况，可以使用 AbstractRoutingConnectionFactory#setConsistentConfirmsReturns 关闭该校验。注意，添加到 AbstractRoutingConnectionFactory 的第一个连接工厂将决定 confirms 和 returns 的总体值。

如果你遇到某些消息希望检查确认/退回、而另一些不希望的情况，这可能有用。例如：

@Bean
public RabbitTemplate rabbitTemplate() {
    final com.rabbitmq.client.ConnectionFactory cf = new com.rabbitmq.client.ConnectionFactory();
    cf.setHost("localhost");
    cf.setPort(5672);

    CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory(cf);
    cachingConnectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);

    PooledChannelConnectionFactory pooledChannelConnectionFactory = new PooledChannelConnectionFactory(cf);

    final Map<Object, ConnectionFactory> connectionFactoryMap = new HashMap<>(2);
    connectionFactoryMap.put("true", cachingConnectionFactory);
    connectionFactoryMap.put("false", pooledChannelConnectionFactory);

    final AbstractRoutingConnectionFactory routingConnectionFactory = new SimpleRoutingConnectionFactory();
    routingConnectionFactory.setConsistentConfirmsReturns(false);
    routingConnectionFactory.setDefaultTargetConnectionFactory(pooledChannelConnectionFactory);
    routingConnectionFactory.setTargetConnectionFactories(connectionFactoryMap);

    final RabbitTemplate rabbitTemplate = new RabbitTemplate(routingConnectionFactory);

    final Expression sendExpression = new SpelExpressionParser().parseExpression(
            "messageProperties.headers['x-use-publisher-confirms'] ?: false");
    rabbitTemplate.setSendConnectionFactorySelectorExpression(sendExpression);
}

这样，带 x-use-publisher-confirms: true 头的消息会通过缓存连接发送，你可以确保消息投递。关于确保消息投递的更多信息，参见发布确认与退回。

队列亲和与 LocalizedQueueConnectionFactory

在集群中使用 HA 队列时，为获得最佳性能，你可能希望连接到主队列所在的物理 broker。CachingConnectionFactory 可以配置多个 broker 地址。这是为了故障转移，客户端按照配置的 AddressShuffleMode 顺序尝试连接。LocalizedQueueConnectionFactory 使用管理插件提供的 REST API 确定哪个节点是该队列的主节点。然后它创建（或从缓存取出）一个只连接到该节点的 CachingConnectionFactory。如果连接失败，会确定新的主节点，消费者连接到它。LocalizedQueueConnectionFactory 配置有一个默认连接工厂，以防无法确定队列的物理位置，此时它照常连接到集群。

LocalizedQueueConnectionFactory 是一个 RoutingConnectionFactory，SimpleMessageListenerContainer 使用队列名作为查找键，如上面路由连接工厂所述。

备注

正因如此（使用队列名查找），LocalizedQueueConnectionFactory 只能在容器配置为监听单个队列时使用。

备注

RabbitMQ 管理插件必须在每个节点上启用。

警告

此连接工厂面向长生命周期连接，例如 SimpleMessageListenerContainer 使用的连接。它不适合短连接使用（例如 RabbitTemplate），因为建立连接前调用 REST API 有开销。此外，对于发布操作，队列是未知的，消息反正会发布到所有集群节点，因此查找节点的逻辑几乎没有价值。

以下示例配置展示了如何配置这些工厂：

@Autowired
private ConfigurationProperties props;

@Bean
public CachingConnectionFactory 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 LocalizedQueueConnectionFactory queueAffinityCF(
        @Qualifier("defaultConnectionFactory") ConnectionFactory defaultCF) {
    return new LocalizedQueueConnectionFactory(defaultCF,
            StringUtils.commaDelimitedListToStringArray(this.props.getAddresses()),
            StringUtils.commaDelimitedListToStringArray(this.props.getAdminUris()),
            StringUtils.commaDelimitedListToStringArray(this.props.getNodes()),
            this.props.getVirtualHost(), this.props.getUsername(), this.props.getPassword(),
            false, null);
}

注意，前三个参数是 addresses、adminUris 和 nodes 的数组。它们是按位置对应的：当容器尝试连接到队列时，它使用 admin API 确定哪个节点是该队列的主节点，并连接到与该节点数组位置相同的地址。

important

从 3.0 版本开始，访问 Rest API 不再使用 RabbitMQ http-client。相反，如果类路径上有 spring-webflux，默认使用 Spring Webflux 的 WebClient；否则使用 RestTemplate。

要把 WebFlux 添加到类路径：

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

implementation 'org.springframework.amqp:spring-rabbit'

你也可以通过实现 LocalizedQueueConnectionFactory#setNodeLocator 并覆盖其 createClient、restCall 以及可选的 close 方法，来使用其他 REST 技术。

lqcf.setNodeLocator(new NodeLocator<MyClient>() {

    @Override
    public MyClient createClient(String userName, String password) {
        ...
    }

    @Override
    public Map<String, Object> restCall(MyClient client, String baseUri, String vhost, String queue) throws URISyntaxException {
        ...
    }

});

框架提供了 WebFluxNodeLocator 和 RestTemplateNodeLocator，默认如上所述。

发布确认与退回

通过把 CachingConnectionFactory 属性 publisherConfirmType 设置为 ConfirmType.CORRELATED、把 publisherReturns 属性设置为 'true'，支持已确认（带关联）和已退回的消息。

设置这些选项后，工厂创建的 Channel 实例会被包装在 PublisherCallbackChannel 中，用于便利回调。获取此类 channel 时，客户端可以向 Channel 注册 PublisherCallbackChannel.Listener。PublisherCallbackChannel 实现包含把确认或退回路由到相应监听器的逻辑。这些特性在下面几节进一步解释。

另请参见关联的发布确认与退回以及作用域操作中的 simplePublisherConfirms。

提示

更多背景信息，参见 RabbitMQ 团队题为介绍发布确认的博客文章。

连接和 Channel 监听器

连接工厂支持注册 ConnectionListener 和 ChannelListener 实现。这让你可以接收连接和 channel 相关事件的通知。（RabbitAdmin 使用 ConnectionListener 在连接建立时执行声明——更多信息参见交换机、队列和绑定的自动声明）。下面的清单展示了 ConnectionListener 接口定义：

@FunctionalInterface
public interface ConnectionListener {

    void onCreate(Connection connection);

    default void onClose(Connection connection) {
    }

    default void onShutDown(ShutdownSignalException signal) {
    }

}

从 2.0 版本开始，可以为 org.springframework.amqp.rabbit.connection.Connection 对象提供 com.rabbitmq.client.BlockedListener 实例，以接收连接阻塞和解除阻塞事件的通知。下例展示了 ChannelListener 接口定义：

@FunctionalInterface
public interface ChannelListener {

    void onCreate(Channel channel, boolean transactional);

    default void onShutDown(ShutdownSignalException signal) {
    }

}

关于你可能希望注册 ChannelListener 的一个场景，参见发布是异步的——如何检测成功与失败。

记录 Channel 关闭事件

1.5 版本引入了一种机制，让用户可以控制日志级别。

AbstractConnectionFactory 使用默认策略记录 channel 关闭，如下所示：

• 正常 channel 关闭（200 OK）不记录日志。
• 如果 channel 因被动队列声明失败而关闭，以 DEBUG 级别记录。
• 如果 channel 因独占消费者条件导致 basic.consume 被拒绝而关闭，以 DEBUG 级别记录（3.1 起，以前为 INFO）。
• 其他所有情况以 ERROR 级别记录。

要修改此行为，可以在 CachingConnectionFactory 的 closeExceptionLogger 属性中注入自定义的 ConditionalExceptionLogger。

此外，AbstractConnectionFactory.DefaultChannelCloseLogger 现在是 public 的，允许被子类化。

另请参见消费者事件。

运行时缓存属性

从 1.6 版本开始，CachingConnectionFactory 现在通过 getCacheProperties() 方法提供缓存统计。这些统计可用于调整缓存，在生产环境中优化它。例如，高水位线可用于确定是否应增大缓存大小。如果它等于缓存大小，你可能希望考虑进一步增大。下表描述了 CacheMode.CHANNEL 属性：

表 1. CacheMode.CHANNEL 的缓存属性

属性	含义
connectionName	由 ConnectionNameStrategy 生成的连接名。
channelCacheSize	当前配置的允许空闲的最大 channel 数。
localPort	连接的本地端口（如果可用）。可用于在 RabbitMQ Admin UI 上与连接和 channel 关联。
idleChannelsTx	当前空闲（缓存）的事务性 channel 数。
idleChannelsNotTx	当前空闲（缓存）的非事务性 channel 数。
idleChannelsTxHighWater	曾并发空闲（缓存）的事务性 channel 最大数。
idleChannelsNotTxHighWater	曾并发空闲（缓存）的非事务性 channel 最大数。

下表描述了 CacheMode.CONNECTION 属性：

表 2. CacheMode.CONNECTION 的缓存属性

属性	含义
connectionName:<localPort>	由 ConnectionNameStrategy 生成的连接名。
openConnections	表示到 broker 连接的连接对象数。
channelCacheSize	当前配置的允许空闲的最大 channel 数。
connectionCacheSize	当前配置的允许空闲的最大连接数。
idleConnections	当前空闲的连接数。
idleConnectionsHighWater	曾并发空闲的连接最大数。
idleChannelsTx:<localPort>	此连接当前空闲（缓存）的事务性 channel 数。你可以使用属性名的 localPort 部分在 RabbitMQ Admin UI 上与连接和 channel 关联。
idleChannelsNotTx:<localPort>	此连接当前空闲（缓存）的非事务性 channel 数。属性名的 localPort 部分可用于在 RabbitMQ Admin UI 上与连接和 channel 关联。
idleChannelsTxHighWater:<localPort>	曾并发空闲（缓存）的事务性 channel 最大数。localPort 部分可用于在 RabbitMQ Admin UI 上与连接和 channel 关联。
idleChannelsNotTxHighWater:<localPort>	曾并发空闲（缓存）的非事务性 channel 最大数。你可以使用属性名的 localPort 部分在 RabbitMQ Admin UI 上与连接和 channel 关联。

还包含 cacheMode 属性（CHANNEL 或 CONNECTION）。

图 1. JVisualVM 示例

RabbitMQ 自动连接/拓扑恢复

从 Spring AMQP 第一个版本起，框架就在 broker 故障时提供自己的连接和 channel 恢复。此外，如配置 Broker 所述，连接重新建立时 RabbitAdmin 会重新声明任何基础设施 bean（队列等）。因此它不依赖 amqp-client 库现在提供的自动恢复。amqp-client 默认启用自动恢复。两种恢复机制之间存在一些不兼容，因此默认情况下，Spring 把底层 RabbitMQ connectionFactory 上的 automaticRecoveryEnabled 属性设置为 false。即使该属性为 true，Spring 也会通过立即关闭任何已恢复连接来有效禁用它。

important

默认情况下，只有定义为 bean 的元素（队列、交换机、绑定）会在连接失败后重新声明。如何更改此行为，参见恢复自动删除声明。