RabbitMQ-Spring AMQP【翻译】4.1.2 Reference——连接和资源管理

57 阅读 0 评论 38 点赞

我是靠谱客的博主酷酷方盒，最近开发中收集的这篇文章主要介绍RabbitMQ-Spring AMQP【翻译】4.1.2 Reference——连接和资源管理，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

版本：2.2.4.RELEASE
地址：官方文档

文章目录

- - 4.1.2 连接和资源管理

4.1.2 连接和资源管理

虽然我们在上一节中描述的AMQP模型是通用的，并且适用于所有实现，但当我们进入资源管理时，细节是特定于代理实现的.因此在本节中，我们关注于spring-rabbit中的代码，此时rabbitmq是唯一被支持的实现。

用于管理RabbitMQ broker连接的中心组件是ConnectionFactory接口。ConnectionFactory实现的职责是提供一个org.springframework.amqp.rabbit.connection.Connection的实例，这个实例时org.springframework.amqp.rabbit.connection.Connection的包装。我们提供的唯一具体实现是CachingConnectionFactory，默认情况下，它建立一个可由应用程序共享的连接代理。共享连接是可能的，因为与AMQP进行消息传递的“工作单元”实际上是一个“通道”（在某些方面，这类似于JMS中连接和会话之间的关系）。这个connection实例提供了一个createChannel()方法。CachingConnectionFactory实现支持缓存这些channels，并且它根据channel是否是事务性的，为channel维护单独的缓存。当创建CachingConnectionFactory实例时，你需要提供给构造方法hostname，你还需要体哦那个username和password属性。要配置通道缓存的大小（默认值为25），可以调用setChannelCacheSize()方法。

从1.3版本开始，你可以将CachingConnectionFactory设置为缓存连接以及仅缓存通道。在这种情况下，对createConnection()的每次调用都会创建一个新连接（或从缓存中取一个空闲连接）。关闭连接会将其返回到缓存（如果尚未达到缓存大小）。在此类连接上创建的通道也会被缓存。在某些情况，使用单独的连接可能很有用，例如使用HA集群中的连接，结合负载均衡，连接到不同的集群成员，等等。要缓存连接，请将cacheMode设置为CacheMode.CONNECTION。
这并不限制连接的数量。相反，它指定允许多少空闲的打开连接。从版本1.5.5开始，提供了一个名为connectionLimit的新属性。设置此属性时，它会限制允许的连接总数。设置后，如果达到限制，则使用channelCheckoutTimeLimit等待连接变为空闲。如果超过该时间，将抛出AmqpTimeoutException。
当cacheMode是CONNECTION时，不支持自动声明队列、交换机和绑定。此外，在撰写本文时，amqp-client库默认情况下会为每个连接创建一个固定的线程池（默认线程数量：Runtime.getRuntime().availableProcessors() * 2）。当使用大量连接时，你应该考虑设置一个自定义的executor在CachingConnectionFactory上。然后，这个executor和它的线程就可以被所有连接共享。执行器的线程池应该是无边界的，或者根据预期用途进行适当设置（通常，每个连接至少有一个线程）。
如果在每个连接上创建多个通道，则池大小会影响并发性，因此变量（或简单缓存）线程池执行器最合适。

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

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

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

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

下面一些例子用来展示如何创建链接：

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>

此外，使用名称空间，您可以添加“通道缓存大小”属性，如下所示：

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

默认缓存模式为通道，但您可以将其配置为缓存连接。在以下示例中，我们使用连接缓存大小：

<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" shuffle-addresses="true"/>

想要了解更多的关于shuffle-addresses信息请看：Connecting to a Cluster

下面是一个自定义线程工厂的示例，该工厂使用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>