连接和资源管理

虽然我们在上一节中描述的 AMQP 模型是通用的，适用于所有实现，但当涉及到资源管理时，细节就特定于代理实现。因此，在本节中，我们将重点关注仅存在于我们的“spring-rabbit”模块中的代码，因为目前 RabbitMQ 是唯一受支持的实现。

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

选择连接工厂

有三个连接工厂可供选择

  • PooledChannelConnectionFactory

  • ThreadChannelConnectionFactory

  • CachingConnectionFactory

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

对于大多数用例，应使用 CachingConnectionFactory。如果您希望确保严格的消息排序而无需使用 作用域操作，则可以使用 ThreadChannelConnectionFactory。PooledChannelConnectionFactory 与 CachingConnectionFactory 类似，因为它使用单个连接和一个通道池。它的实现更简单，但它不支持相关发布者确认。

所有三个工厂都支持简单的发布者确认。

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

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

PooledChannelConnectionFactory

此工厂管理单个连接和两个通道池，基于 Apache Pool2。一个池用于事务性通道，另一个池用于非事务性通道。池是 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;
}

ThreadChannelConnectionFactory

此工厂管理单个连接和两个 ThreadLocal，一个用于事务性通道，另一个用于非事务性通道。此工厂确保同一线程上的所有操作都使用相同的通道（只要它保持打开状态）。这有助于在无需 作用域操作 的情况下实现严格的消息排序。为避免内存泄漏，如果您的应用程序使用许多短暂的线程，则必须调用工厂的 closeThreadChannel() 来释放通道资源。从 2.3.7 版本开始，线程可以将其通道传输到另一个线程。有关更多信息，请参阅 多线程环境中的严格消息排序。

CachingConnectionFactory

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

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

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

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

当缓存模式为CONNECTION时，不支持队列和其他对象的自动声明（请参阅交换机、队列和绑定的自动声明）。

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

重要的是要了解缓存大小（默认情况下）不是限制，而仅仅是可以缓存的通道数量。如果缓存大小为 10，则实际上可以有任意数量的通道正在使用。如果正在使用超过 10 个通道并且它们都返回到缓存，则 10 个进入缓存。其余的将被物理关闭。

从 1.6 版开始，默认通道缓存大小已从 1 增加到 25。在高容量的多线程环境中，较小的缓存意味着通道以较高的速率创建和关闭。增加默认缓存大小可以避免此开销。您应该通过 RabbitMQ 管理 UI 监控正在使用的通道，如果看到许多通道正在创建和关闭，则考虑进一步增加缓存大小。缓存仅按需增长（以满足应用程序的并发需求），因此此更改不会影响现有的低容量应用程序。

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

框架内使用的通道（例如，RabbitTemplate）可靠地返回到缓存。如果您在框架外部创建通道（例如，通过直接访问连接并调用createChannel()），则必须可靠地返回它们（通过关闭），可能在finally块中，以避免用完通道。

以下示例显示了如何创建一个新的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更简单，因为它不缓存通道，但由于其性能和弹性不足，因此不适合在简单的测试之外实际使用。如果您需要出于某种原因实现自己的ConnectionFactory，则AbstractConnectionFactory基类可能提供一个不错的起点。

可以使用 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"/>

或者，如果在集群环境中运行，则可以使用 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即可。Boot 会自动检测该 bean 并将其连接到工厂。

阻塞的连接和资源限制

连接可能会被对应于内存警报的代理阻止交互。从 2.0 版开始，可以为org.springframework.amqp.rabbit.connection.Connection提供com.rabbitmq.client.BlockedListener实例，以接收连接阻塞和解除阻塞事件的通知。此外，AbstractConnectionFactory分别通过其内部BlockedListener实现发出ConnectionBlockedEvent和ConnectionUnblockedEvent。这些允许您提供应用程序逻辑以对代理上的问题做出适当的反应，例如采取一些纠正措施。

当应用程序配置为使用单个CachingConnectionFactory时（默认情况下，Spring Boot 自动配置就是这样），当连接被代理阻塞时，应用程序将停止工作。当它被代理阻塞时，其任何客户端都将停止工作。如果我们在同一个应用程序中拥有生产者和消费者，当生产者阻塞连接（因为代理上没有更多资源）并且消费者无法释放它们（因为连接被阻塞）时，我们最终可能会出现死锁。为了缓解这个问题，我们建议再创建一个单独的CachingConnectionFactory实例，并使用相同的选项——一个用于生产者，一个用于消费者。对于在消费者线程上执行的事务性生产者，单独的CachingConnectionFactory是不可能的，因为它们应该重用与消费者事务关联的Channel。

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

从 1.7.7 版开始，提供了一个AmqpResourceNotAvailableException，当SimpleConnection.createChannel()无法创建Channel时（例如，因为channelMax限制已达到并且缓存中没有可用的通道），就会抛出此异常。您可以在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 自动恢复，以避免在代理可用但连接尚未恢复时出现 AutoRecoverConnectionNotCurrentlyOpenException 实例。例如，当在 RabbitTemplate 中配置 RetryTemplate 时，即使故障转移到集群中的另一个代理，您也可能会注意到此异常。由于自动恢复连接按计时器恢复，因此使用 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;
}
Boot application.properties
spring.rabbitmq.ssl.enabled:true
spring.rabbitmq.ssl.keyStore=...
spring.rabbitmq.ssl.keyStoreType=jks
spring.rabbitmq.ssl.keyStorePassword=...
spring.rabbitmq.ssl.trustStore=...
spring.rabbitmq.ssl.trustStoreType=jks
spring.rabbitmq.ssl.trustStorePassword=...
spring.rabbitmq.host=...
...
XML
<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="classpath:secrets/rabbitSSL.properties"/>
</bean>

有关配置 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 Resources。通常，此属性文件由操作系统通过应用程序具有读取访问权限来保护。

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

从 2.0 版本开始，服务器证书默认情况下会进行验证，因为这样更安全。如果您出于某种原因希望跳过此验证，请将工厂 Bean 的 skipServerCertificateValidation 属性设置为 true。从 2.1 版本开始，RabbitConnectionFactoryBean 现在默认情况下会调用 enableHostnameVerification()。要恢复到之前的行为，请将 enableHostnameVerification 属性设置为 false。
从 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]。

目标（如果提供，则为默认）连接工厂必须具有发布者确认和返回的相同设置。请参阅发布者确认和返回。

从 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 队列时，为了获得最佳性能，您可能希望连接到驻留主队列的物理代理。CachingConnectionFactory 可以配置多个代理地址。这是为了进行故障转移，客户端会根据配置的 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 来确定哪个节点是队列的主节点，并连接到与该节点在同一数组位置中的地址。

从 3.0 版本开始，不再使用 RabbitMQ 的 http-client 来访问 Rest API。取而代之，默认情况下，如果类路径中存在 spring-webflux，则使用 Spring Webflux 中的 WebClient；否则使用 RestTemplate。

将 WebFlux 添加到类路径中

Maven
<dependency>
  <groupId>org.springframework.amqp</groupId>
  <artifactId>spring-rabbit</artifactId>
</dependency>
Gradle
compile 'org.springframework.amqp:spring-rabbit'

您还可以通过实现 LocalizedQueueConnectionFactory.NodeLocator 并覆盖其 createClient、restCall 以及可选的 close 方法来使用其他 REST 技术。

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

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

    @Override
    public HashMap<String, Object> restCall(MyClient client, URI uri) {
        ...
    });

});

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

发布者确认和返回

通过将 CachingConnectionFactory 属性 publisherConfirmType 设置为 ConfirmType.CORRELATED 以及 publisherReturns 属性设置为 'true'，可以支持确认（带相关性）和返回的消息。

当这些选项设置后，工厂创建的 Channel 实例会被包装在一个 PublisherCallbackChannel 中，用于促进回调。当获取到这样的通道时，客户端可以在 Channel 上注册一个 PublisherCallbackChannel.Listener。PublisherCallbackChannel 实现包含将确认或返回路由到相应侦听器的逻辑。这些功能将在以下部分中进一步解释。

另请参阅 相关发布者确认和返回 以及 作用域操作 中的 simplePublisherConfirms。

有关更多背景信息，请参阅 RabbitMQ 团队发布的博文 Introducing Publisher Confirms。

连接和通道侦听器

连接工厂支持注册 ConnectionListener 和 ChannelListener 实现。这允许您接收与连接和通道相关的事件通知。（ConnectionListener 由 RabbitAdmin 用于在建立连接时执行声明 - 有关更多信息，请参阅 交换机、队列和绑定的自动声明）。以下清单显示了 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 的一种场景，请参阅 发布是异步的——如何检测成功和失败。

记录通道关闭事件

1.5 版本引入了一种机制，允许用户控制日志级别。

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

  • 正常的通道关闭（200 OK）不会记录。

  • 如果通道由于被动队列声明失败而关闭，则会在 DEBUG 级别记录。

  • 如果通道关闭是因为 basic.consume 由于排他性消费者条件而被拒绝，则会在 DEBUG 级别记录（从 3.1 开始，之前为 INFO）。

  • 所有其他情况都会在 ERROR 级别记录。

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

此外，AbstractConnectionFactory.DefaultChannelCloseLogger 现在是公开的，允许对其进行子类化。

另请参阅 消费者事件。

运行时缓存属性

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

表 1. CacheMode.CHANNEL 的缓存属性
属性 含义 
connectionName

由 ConnectionNameStrategy 生成的连接的名称。

 
channelCacheSize

当前配置的允许空闲的最大通道数。

 
localPort

连接的本地端口（如果可用）。这可用于在 RabbitMQ Admin UI 上与连接和通道相关联。

 
idleChannelsTx

当前空闲（缓存）的事务性通道数。

 
idleChannelsNotTx

当前空闲（缓存）的非事务性通道数。

 
idleChannelsTxHighWater

并发空闲（缓存）的事务性通道的最大数量。

 
idleChannelsNotTxHighWater

并发空闲（缓存）的非事务性通道的最大数量。

下表描述了 CacheMode.CONNECTION 属性

表 2. CacheMode.CONNECTION 的缓存属性
属性 含义 
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）。

cacheStats
图 1. JVisualVM 示例

RabbitMQ 自动连接/拓扑恢复

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

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