上海网站设计合理柚v米科技,网站关键词描述,专业做网站建设的,给别人做网站怎么赚钱吗转自#xff1a;
【RabbitMQ-3】连接池的配置_小胖学编程的博客-CSDN博客文章目录1. rabbitmq的connection连接池1.1 问题提出1.1.1 Connection对象管理以及性能1.1.2 Channel对象管理以及性能1.2 Spring AMQP线程池配置1.2.1 ConnectionFactory连接工厂1.2.2 消费发送和接收…转自
【RabbitMQ-3】连接池的配置_小胖学编程的博客-CSDN博客文章目录1. rabbitmq的connection连接池1.1 问题提出1.1.1 Connection对象管理以及性能1.1.2 Channel对象管理以及性能1.2 Spring AMQP线程池配置1.2.1 ConnectionFactory连接工厂1.2.2 消费发送和接收使用不同的Connectionjava NIO是IO的多路复用Channel连接是TCP的多路复用。那么他们有什么关系呢NIO是服务器开启一个线程在内核中使用select()进行轮询管理一些socket当sockehttps://blog.csdn.net/qq_29595463/article/details/107858293 文章目录 1. rabbitmq的connection连接池 1.1 问题提出 1.1.1 Connection对象管理以及性能 1.1.2 Channel对象管理以及性能 1.2 Spring AMQP线程池配置 1.2.1 ConnectionFactory连接工厂 1.2.2 消费发送和接收使用不同的Connection java NIO是IO的多路复用Channel连接是TCP的多路复用。那么他们有什么关系呢 NIO是服务器开启一个线程在内核中使用select()进行轮询管理一些socket当socket数据准备好时会通知应用程序进行读写请求。系统之间那点事-NIO内附IO模型-IO/NIO/AIO到底是什么。服务器看起来就好像是一个socket在通信实现了多路复用。 channel复用TCP连接是为了避免TCP连接创建和销毁的性能损耗而多个channel使用一个tcp连接。
1. rabbitmq的connection连接池
RabbitMQ.png
我们看到一个Connection里面可以包含多个channel。那么我们在连接broker时connection和channel的关系是什么 1.1 问题提出 1.1.1 Connection对象管理以及性能
Connection连接本质上就是TCP连接系统之间那点事-问题驱动-TCP的连接和关闭是比较耗费时间的。我们可以使用一个单例的Connection对象创建多个Channel来实现数据传输但是在channel信息比较大的情况下Connection带宽会限制消息的传输。那么需要设计Connection池将流量分摊到不同的connection上。 1.1.2 Channel对象管理以及性能
Channel对象的创建和销毁也是非常耗时的推荐共享使用Channel而不是每次都创建和销毁Channel。那如何设计一个channel线程池呢
官网对于Connection的解读 AMQP 0-9-1 connections are typically long-lived. AMQP 0-9-1 is an application level protocol that uses TCP for reliable delivery. Connections use authentication and can be protected using TLS. When an application no longer needs to be connected to the server, it should gracefully close its AMQP 0-9-1 connection instead of abruptly closing the underlying TCP connection.
大概意思就是AMQP 0-9-1一般是一个TCP的长链接当应用程序不再需要连接到服务器时应该正常关闭AMQP 0-9-1连接而不是关闭TCP连接。
官网对于Channel的解读 Some applications need multiple connections to the broker. However, it is undesirable to keep many TCP connections open at the same time because doing so consumes system resources and makes it more difficult to configure firewalls. AMQP 0-9-1 connections are multiplexed withchannels that can be thought of as “lightweight connections that share a single TCP connection”. Every protocol operation performed by a client happens on a channel. Communication on a particular channel is completely separate from communication on another channel, therefore every protocol method also carries a channel ID (a.k.a. channel number), an integer that both the broker and clients use to figure out which channel the method is for. A channel only exists in the context of a connection and never on its own. When a connection is closed, so are all channels on it. For applications that use multiple threads/processes for processing, it is very common to open a new channel per thread/process and not share channels between them.
一些应用需要同时创建多个连接到broker也就是RabbitMQ服务器上。然而因为防火墙的存在很难同时创建多个连接。 AMQP 0-9-1连接使用多个channel连接实现对单一Connection的复用。 客户端的每一个协议操作都发送在channel上。每个协议方法携带者channel ID。broker和client使用channel ID来确定方法对应的channel。因此实现channel之间的数据隔离。 channel不能单独存在仅存在connection上下文中。当connection关闭时channel也会关闭。 多线程/进程之间打开一个channel但不共享channels是很普遍的。
通道和并发注意事项线程安全 As a rule of thumb, sharing Channel instances between threads is something to be avoided. Applications should prefer using a Channel per thread instead of sharing the same Channel across multiple threads.
线程之间共享channel是无法避免的应用程序跟喜欢每个线程使用一个channel而不是跨线程共享相同的channel。 A classic anti-pattern to be avoided is opening a channel for each published message. Channels are supposed to be reasonably long-lived and opening a new one is a network round-trip which makes this pattern extremely inefficient.
要避免一个反例为每一个发布的消息分配一个channel开辟一个新的channel需要一个网络的往返这种模式是很低效的。channel保持合理的存活时间。 It is possible to use channel pooling to avoid concurrent publishing on a shared channel: once a thread is done working with a channel, it returns it to the pool, making the channel available for another thread. Channel pooling can be thought of as a specific synchronization solution. It is recommended that an existing pooling library is used instead of a homegrown solution. For example, Spring AMQP which comes with a ready-to-use channel pooling feature.
敲黑板划重点可以使用channel pool来避免共享channel上并发发布一旦一个线程使用完了channel那么它将返回到pool中。其他线程便可使用这个Channel。线程池是一个解决方案可以使用 Spring AMQP线程池而不是自己开发。
__总结__频繁建立TCP连接和channel连接是消耗性能的于是我们希望可以共享connection或者channel。达到连接的复用。 1.2 Spring AMQP线程池配置
版本spring-rabbit:2.0.2.RELEASE 1.2.1 ConnectionFactory连接工厂
这个ConnectionFactory是Spring AMQP定义的连接工厂负责创建连接。而CacheConnectionFactory实现支持对这些通道的缓存。 private static ConnectionFactory newRabbitConnectionFactory() { ConnectionFactory connectionFactory new ConnectionFactory(); connectionFactory.setAutomaticRecoveryEnabled(false); return connectionFactory; } 1 2 3 4 5
参数分析
1. 开启confirm机制。 connectionFactory.setPublisherConfirms(true); connectionFactory.setPublisherReturns(true); 为了消息的不丢失生产者可以设置事务或者confirm异步通知。但是事务性能并不是很好所以一般使用confirm模式。 区别confirm保证达到交换机return保证交换机到达队列 如果消息没有到exchange,则confirm回调,ackfalse 如果消息到达exchange,则confirm回调,acktrue exchange到queue成功,则不回调return exchange到queue失败,则回调return(需设置mandatorytrue,否则不回回调,消息就丢了)
*注意设置PublisherReturns状态为true那么需要设置 rabbitTemplate.setMandatory(true); 具体如何保证消息不丢失请参考RabbitMQ的消息不丢失机制
2. 配置模式 缓存模式一般两种 connectionFactory.setCacheMode(CachingConnectionFactory.CacheMode.CONNECTION); public static enum CacheMode { CHANNEL, CONNECTION; private CacheMode() { } } 1 2 3 4 5 6 7
2.1 CHANNEL模式
程序运行期间ConnectionFactory只维护着一个connection但是可以含有多个channel操作rabbitmq之前必须先获取一个channel否则将会阻塞。
相关参数配置
connectionFactory.setChannelCacheSize(10); 设置每个Connection中的缓存Channel的数量。操作rabbitmq之前send/receive message等要先获取到一个Channel获取Channel时会先从缓存中找闲置的Channel如果没有则创建新的Channel当Channel数量大于缓存数量时多出来没法放进缓存的会被关闭。
connectionFactory.setChannelCheckoutTimeout(600); 单位毫秒当这个值大于0时ChannelCacheSize代表的是缓存的数量上限当缓存获取不到可用的channel时不会创建新的channel会等待指定的时间若到时间后还获取不到可用的channel直接抛出AmqpTimeoutException。
注意在CONNECTION模式这个值也会影响获取Connection的等待时间超时获取不到Connection也会抛出AmqpTimeoutException异常。
2.2 CONNECTION模式
CONNECTION模式。在这个模式下允许创建多个connection会缓存一定数量的connection每个connection中同样缓存着一些channel。
相关参数配置
connectionFactory.setConnectionCacheSize(3); 仅在CONNECTION模式下使用指定connection缓存数量。
connectionFactory.setConnectionLimit(10); 仅在CONNECTION模式下使用指定connection数量上限。
官网对于是否关闭channel解答 Channels used within the framework (e.g. RabbitTemplate) will be reliably returned to the cache. If you create channels outside of the framework, (e.g. by accessing the connection(s) directly and invoking createChannel()), you must return them (by closing) reliably, perhaps in a finally block, to avoid running out of channels.
注意若使用RabbitTemplate创建channel那么无需关闭但是自己新建connection创建channel则需要手动关闭避免channel溢出。
ConnectionFactory 代码 Bean public ConnectionFactory connectionFactory() { CachingConnectionFactory connectionFactory new CachingConnectionFactory(host, port); connectionFactory.setUsername(username); connectionFactory.setPassword(password); //设置virtualHost。 connectionFactory.setVirtualHost(/); //消息的确认机制confirm connectionFactory.setPublisherConfirms(true); connectionFactory.setPublisherReturns(true); //setCacheMode设置缓存模式共有两种CHANNEL和CONNECTION模式。 //1、CONNECTION模式这个模式下允许创建多个Connection会缓存一定数量的Connection每个Connection中同样会缓存一些Channel // 除了可以有多个Connection其它都跟CHANNEL模式一样。 //2、CHANNEL模式程序运行期间ConnectionFactory会维护着一个Connection // 所有的操作都会使用这个Connection但一个Connection中可以有多个Channel // 操作rabbitmq之前都必须先获取到一个Channel // 否则就会阻塞可以通过setChannelCheckoutTimeout()设置等待时间 // 这些Channel会被缓存缓存的数量可以通过setChannelCacheSize()设置 connectionFactory.setCacheMode(CachingConnectionFactory.CacheMode.CONNECTION); //设置CONNECTION模式可创建多个Connection连接 //设置每个Connection中缓存Channel的数量不是最大的。操作rabbitmq之前send/receive message等 // 要先获取到一个Channel.获取Channel时会先从缓存中找闲置的Channel如果没有则创建新的Channel // 当Channel数量大于缓存数量时多出来没法放进缓存的会被关闭。 connectionFactory.setChannelCacheSize(10); //单位毫秒配合channelCacheSize不仅是缓存数量而且是最大的数量。 // 从缓存获取不到可用的Channel时不会创建新的Channel会等待这个值设置的毫秒数 //同时在CONNECTION模式这个值也会影响获取Connection的等待时间 // 超时获取不到Connection也会抛出AmqpTimeoutException异常。 connectionFactory.setChannelCheckoutTimeout(600); //仅在CONNECTION模式使用设置Connection的缓存数量。 connectionFactory.setConnectionCacheSize(3); //setConnectionLimit仅在CONNECTION模式使用设置Connection的数量上限。 connectionFactory.setConnectionLimit(10); return connectionFactory; } 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35
RabbitTemplate 代码 Autowired Bean public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) { //客户端开启confirm模式 RabbitTemplate rabbitTemplate new RabbitTemplate(connectionFactory); rabbitTemplate.setMandatory(true); rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() { Override public void confirm(CorrelationData correlationData, boolean ack, String cause) { log.info(消息发送成功:correlationData({}),ack({}),cause({}), correlationData, ack, cause); } }); rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() { Override public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) { log.info(消息丢失:exchange({}),route({}),replyCode({}),replyText({}),message:{}, exchange, routingKey, replyCode, replyText, message); } }); return rabbitTemplate; } 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20
1.2.2 消费发送和接收使用不同的Connection 当一个服务同时作为消息的发送端和接收端时建议使用不同的Connection避免一方出现故障或者阻塞影响另一方。
只需要在RabbitTemplate中加入下面的配置那么RabbitTemplate在创建Connection时会根据这个boolean的值选择使用ConnectionFactory本身或者ConnectionFactory中的publisherConnectionFactory也即是一个ConnectionFactory来创建。
rabbitTemplate.setUsePublisherConnection(true);
