在 Netty 里,有4个方法用来查询 Channel 的状态:isOpen,isRegistered,isActive,isWritable,其中,isWritable 在并发量很高时会返回很多 false。
isWritable 是什么含义?
isWritable:Returns true if and only if the I/O thread will perform the requested write operation immediately. Any write requests made when this method returns false are queued until the I/O thread is ready to process the queued write requests.
为什么 isWritable 会返回 false?而且这个问题在 stackoverflow,netty 社区也被很多人问到。
i.e.
if (channel.isWritable()) {
channel.writeAndFlush(data);
}
首先,说下 netty 处理 writeAndFlush 的原理:
1、业务线程调用 writeAndFlush 发送消息,会生成 WriteAndFlushTask,交由 IO 线程处理,write 操作将消息写入 ChannelOutboundBuffer(不会写到 socket),flush 操作将 ChannelOutboundBuffer 写 入socket 的发送缓冲区;(这里注意,writeAndFlush 它只是一个语法糖,意味着这不是原子操作,因此在此方法执行的中间,可能在多个线程之间进行了上下文切换。)
2、ChannelOutboundBuffer 它配置一个高水位线和低水位线,当 buffer 的大小超过高水位线的时候对应 channel 的 isWritable 就会变成 false,当 buffer 的大小低于低水位线的时候,isWritable 就会变成 true。
其中,高水位线和低水位线是字节数,默认高水位是64K,低水位是32K,通过以下方式进行设置:
.option(ChannelOption.WRITE_BUFFER_HIGH_WATER_MARK, 64 * 1024)
.option(ChannelOption.WRITE_BUFFER_LOW_WATER_MARK, 32 * 1024)
如何解决这个问题?
ChannelOutboundBuffer 的容量过高或过低时都会触发 fireChannelWritabilityChanged 方法,因此可通过重写 channelWritabilityChanged 方法调整消息产生速度。
在常用的中间件里,我们看看它们是如何处理的:
1、Notify 在给订阅组投递消息时,先检查此订阅组的 Channel 是否超过最高位,如果是,则此次不投递,如果不是,继续投递。
2、Flink 核心发送方法中如果 Channel 不可写,则会跳过发送,当 Channel 再次可写后,Netty 会调用该 Handle 的 ChannelWritabilityChanged 方法,从而重新触发发送函数。
3、Duboo 发送请求时,判断是否已经关闭的 Channel,如果是,不再放入连接池,重新申请连接。
总之,在使用 Channel 写数据之前,建议使用 isWritable 方法来判断一下当前 ChannelOutboundBuffer 里的写缓存水位,防止 OOM 发生。