java NIO是IO的多路复用,Channel连接是TCP的多路复用。那么他们有什么关系呢?
NIO
是服务器开启一个线程,在内核中使用select()进行轮询管理一些socket,当socket数据准备好时,会通知应用程序进行读写请求。 系统之间那点事-NIO(内附IO模型)-IO/NIO/AIO到底是什么 。 服务器看起来就好像是一个socket在通信,实现了多路复用。
channel
复用TCP连接,是为了避免TCP连接创建和销毁的性能损耗,而多个channel使用一个tcp连接。1. rabbitmq的connection连接池
我们看到一个
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
线程池呢?1.2 官网解读
官网对于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 with channels 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.3 Spring AMQP线程池配置
版本
spring-rabbit:2.0.2.RELEASE
1.3.1 ConnectionFactory连接工厂
这个
ConnectionFactory
是Spring AMQP定义的连接工厂,负责创建连接。而CacheConnectionFactory
实现支持对这些通道的缓存。private static ConnectionFactory newRabbitConnectionFactory() { ConnectionFactory connectionFactory = new ConnectionFactory(); connectionFactory.setAutomaticRecoveryEnabled(false); return connectionFactory;
参数分析:
1. 开启confirm机制。
connectionFactory.setPublisherConfirms(true); connectionFactory.setPublisherReturns(true);
为了消息的不丢失,生产者可以设置事务或者confirm
异步通知。但是事务性能并不是很好,所以一般使用confirm
模式。
区别:(confirm保证达到交换机,return保证交换机到达队列)
如果消息没有到exchange
,则confirm
回调,ack=false
;
如果消息到达exchange
,则confirm
回调,ack=true
;
exchange
到queue
成功,则不回调return
exchange
到queue
失败,则回调return
(需设置mandatory=true
,否则不回回调,消息就丢了)*注意:设置
PublisherReturns
状态为true
,那么需要设置rabbitTemplate.setMandatory(true);
具体如何保证消息不丢失,请参考RabbitMQ的消息不丢失机制2. 配置模式
缓存模式一般两种:
connectionFactory.setCacheMode(CachingConnectionFactory.CacheMode.CONNECTION);
public static enum CacheMode { CHANNEL, CONNECTION; private CacheMode() {
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;
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;
2.2 消费发送和接收使用不同的Connection