1. 概述
在 netty-handler
模块的 timeout
包,实现 Channel 的读写操作的空闲检测。可能有胖友不太了解空闲检测的具体用途。请先研读理解下 《简易RPC框架-心跳与重连机制》 。
2. 类
timeout
包,包含的类,如下图所示:`timeout` 包
一共有 3 个 ChannelHandler 实现类:
- IdleStateHandler ,当 Channel 的读或者写空闲时间太长时,将会触发一个 IdleStateEvent 事件。然后,你可以自定义一个 ChannelInboundHandler ,重写
#userEventTriggered(ChannelHandlerContext ctx, Object evt)
方法,处理该事件。- ReadTimeoutHandler ,继承 IdleStateHandler 类,当 Channel 的读空闲时间( 读或者写 )太长时,抛出 ReadTimeoutException 异常,并自动关闭该 Channel 。然后,你可以自定一个 ChannelInboundHandler ,重写
#exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
方法,处理该异常。
- ReadTimeoutHandler ,继承 IdleStateHandler 类,当 Channel 的读空闲时间( 读或者写 )太长时,抛出 ReadTimeoutException 异常,并自动关闭该 Channel 。然后,你可以自定一个 ChannelInboundHandler ,重写
- WriteTimeoutHandler ,当一个写操作不能在指定时间内完成时,抛出 WriteTimeoutException 异常,并自动关闭对应 Channel 。然后,你可以自定一个 ChannelInboundHandler ,重写
#exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
方法,处理该异常。
😈 从 WriteTimeoutHandler 可以看出,本文实际不仅仅分享 IdleStateHandler ,更准确的是分享 Timeout 相关的 ChannelHandler 。考虑到大多数胖友对 IdleStateHandler 比较熟悉,也相对常用,所以标题才取了 《精尽 Netty 源码解析 —— ChannelHandler(五)之 IdleStateHandler》 。
3. IdleState
io.netty.handler.timeout.IdleState
,空闲状态枚举。代码如下:
/** |
- 一共有 3 种状态。其中,
ALL_IDLE
表示的是,读或写任一空闲,注意是“或”。
3.1 IdleStateEvent
io.netty.handler.timeout.IdleStateEvent
,空闲事件。代码如下:
public class IdleStateEvent { |
- 3 类(
state
)空闲事件,再组合上是否首次(first
),一共有 6 种空闲事件。
4. TimeoutException
io.netty.handler.timeout.TimeoutException
,继承 ChannelException 类,超时异常。代码如下:
public class TimeoutException extends ChannelException { |
4.1 ReadTimeoutException
io.netty.handler.timeout.ReadTimeoutException
,继承 TimeoutException 类,读超时( 空闲 )异常。代码如下:
public final class ReadTimeoutException extends TimeoutException { |
4.2 WriteTimeoutException
io.netty.handler.timeout.WriteTimeoutException
,继承 TimeoutException 类,写超时( 空闲 )异常。代码如下:
public final class WriteTimeoutException extends TimeoutException { |
5. IdleStateHandler
io.netty.handler.timeout.IdleStateHandler
,继承 ChannelDuplexHandler 类,当 Channel 的读或者写空闲时间太长时,将会触发一个 IdleStateEvent 事件。
5.1 构造方法
老艿艿:高能预警,IdleStateHandler 的属性有点点多。
/** |
- 属性比较多,保持耐心和淡定,我们继续来整理一波。
MIN_TIMEOUT_NANOS
静态属性,最小的超时时间为 1 ,单位:纳秒。因为 IdleStateHandler 创建的,检测定时任务的时间,以纳秒为单位。state
属性,IdleStateHandler 的状态。一共有三种,见注释。- Read 空闲相关属性
readerIdleTimeNanos
属性,配置的读空闲时间,单位:纳秒。readerIdleTimeout
属性,读空闲的定时检测任务。lastReadTime
属性,读空闲的定时检测任务。firstReaderIdleEvent
属性,是否首次读空闲。- 【独有】
reading
属性,是否正在读取。
- Write 空闲相关属性
writerIdleTimeNanos
属性,配置的写空闲时间,单位:纳秒。writerIdleTimeout
属性,写空闲的定时检测任务。lastWriteTime
属性,最后写时间。writeListener
属性,写入操作,完成 flush 到对端的回调监听器。初始时,创建好,避免重复创建,从而减轻 GC 压力。- 【独有】ChannelOutboundBuffer 相关属性
observeOutput
属性, 是否观察 ChannelOutboundBuffer 写入队列。lastChangeCheckTimeStamp
属性,最后检测到 ChannelOutboundBuffer 发生变化的时间。lastMessageHashCode
属性,第一条准备 flash 到对端的消息的 HashCode 。lastPendingWriteBytes
属性,总共等待 flush 到对端的内存大小。- 关于这几个属性,跟着 「5.7 hasOutputChanged」 一起理解。
- ALL 空闲相关属性
- 因为 ALL 是 Write 和 Read 任一,所以共用它们的一些属性。
allIdleTimeNanos
属性,配置的All( 读或写任一 ),单位:纳秒。
5.2 initialize
#initialize(ChannelHandlerContext ctx)
方法,初始化 IdleStateHandler 。代码如下:
1: private void initialize(ChannelHandlerContext ctx) { |
- 第 2 至 9 行:校验状态,避免因为
#destroy()
方法在#initialize(ChannelHandlerContext ctx)
方法,执行之前。 - 第 12 行:标记
state
为已初始化。 第 14 行:调用
#initOutputChanged(ChannelHandlerContext ctx)
方法,初始化 ChannelOutboundBuffer 相关属性。代码如下:private void initOutputChanged(ChannelHandlerContext ctx) {
if (observeOutput) {
Channel channel = ctx.channel();
Unsafe unsafe = channel.unsafe();
ChannelOutboundBuffer buf = unsafe.outboundBuffer();
if (buf != null) {
// 记录第一条准备 flash 到对端的消息的 HashCode
lastMessageHashCode = System.identityHashCode(buf.current());
// 记录总共等待 flush 到对端的内存大小
lastPendingWriteBytes = buf.totalPendingWriteBytes();
}
}
}- 初始化
lastMessageHashCode
和lastPendingWriteBytes
属性。
- 初始化
第 17 至 26 行:根据配置,分别调用
#schedule(hannelHandlerContext ctx, Runnable task, long delay, TimeUnit unit)
方法,初始相应的定时任务。代码如下:ScheduledFuture<?> schedule(ChannelHandlerContext ctx, Runnable task, long delay, TimeUnit unit) {
return ctx.executor().schedule(task, delay, unit);
}
- 一共有 ReaderIdleTimeoutTask、WriterIdleTimeoutTask、AllIdleTimeoutTask 三种任务,下文我们详细解析。
该方法,会在多个 Channel 事件中被调用。代码如下:
// <2> |
<1>
:当客户端与服务端成功建立连接后,Channel 被激活,此时 channelActive 方法,的初始化被调用。<2>
:当 Channel 被激活后,动态添加此 Handler ,则 handlerAdded 方法的初始化被调用。<3>
:当 Channel 被激活后,用户主动切换 Channel 的所在的 EventLoop ,则 channelRegistered 方法的初始化被调用。
5.3 destroy
#destroy()
方法,销毁 IdleStateHandler 。代码如下:
private void destroy() { |
- 标记
state
为已销毁。 - 销毁响应的定时任务。
该方法,会在多个 Channel 事件中被调用。代码如下:
|
5.4 channelIdle
在定时任务中,如果检测到空闲:
① 首先,调用 #newIdleStateEvent(IdleState state, boolean first)
方法,创建对应的空闲事件。代码如下:
protected IdleStateEvent newIdleStateEvent(IdleState state, boolean first) { |
② 然后,调用 #channelIdle(ChannelHandlerContext ctx, IdleStateEvent evt)
方法,在 pipeline 中,触发 UserEvent 事件。代码如下:
/** |
5.5 channelRead
#channelRead(ChannelHandlerContext ctx, Object msg)
方法,代码如下:
|
在开启 read 或 all 的空闲检测的情况下,在【继续传播 Channel Read 事件到下一个节点】之前,会:
- 标记
reading
为正在读取。 - 重置
firstWriterIdleEvent
和firstAllIdleEvent
为true
,即又变成首次。
那么什么时候记录 lastReadTime
最后读取时间呢?答案在 #channelReadComplete(ChannelHandlerContext ctx)
方法中。代码如下:
|
在开启 read 或 all 的空闲检测的情况下,在【继续传播 Channel ReadComplete 事件到下一个节点】之前,会:
记录
lastReadTime
最后读取时间为#ticksInNanos()
方法,代码如下:long ticksInNanos() {
return System.nanoTime();
}- 当前时间,单位:纳秒。
标记
reading
为不在读取。
5.6 write
#write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
方法,代码如下:
|
在开启 write 或 all 的空闲检测的情况下,写入的时候,会添加写入监听器 writeListener
。该监听器会在消息( 数据 ) flush 到对端后,回调,修改最后写入时间 lastWriteTime
为 #ticksInNanos()
。代码如下:
// Not create a new ChannelFutureListener per write operation to reduce GC pressure. |
5.7 hasOutputChanged
老艿艿:关于这个方法,看完 「5.8.2 WriterIdelTimeoutTask」 后,再回过头理解。
#hasOutputChanged(ChannelHandlerContext ctx, boolean first)
方法,判断 ChannelOutboundBuffer 是否发生变化。代码如下:
/** |
- 第 3 行:判断开启观察 ChannelOutboundBuffer 队列。
- 如果
lastChangeCheckTimeStamp
和lastWriteTime
不一样,说明写操作进行过了,则更新此值。 - 第 14 至 17 行:这段逻辑,理论来说不会发生。因为
lastWriteTime
属性,只会在writeListener
回调中修改,那么如果发生lastChangeCheckTimeStamp
和lastWriteTime
不相等,first
必然为true
。因为,Channel 相关的事件逻辑,都在它所在的 EventLoop 中,不会出现并发的情况。关于这一块,基友【莫那一鲁道】在 https://github.com/netty/netty/issues/8251 已经进行提问,坐等结果。
- 如果
- 第 25 至 27 行:获得新的
messageHashCode
和pendingWriteBytes
的。 - 第 29 至 33 行:若发生了变化,则修改最后一次的
lastMessageHashCode
和lastPendingWriteBytes
。messageHashCode != lastMessageHashCode
成立,① 有可能对端接收数据比较慢,导致一个消息发送了一部分;② 又或者,发送的消息非常非常非常大,导致一个消息发送了一部分,就将发送缓存区写满。如果是这种情况下,可以使用 ChunkedWriteHandler ,一条大消息,拆成多条小消息。pendingWriteBytes != lastPendingWriteBytes
成立,① 有新的消息,写到 ChannelOutboundBuffer 内存队列中;② 有几条消息成功写到对端。这种情况,此处不会发生。
- 第 35 至 37 行:当且仅当
first
为true
时,即非首次,才返回true
,表示 ChannelOutboundBuffer 发生变化。- 这是一个有点“神奇”的设定,笔者表示不太理解。理论来说,ChannelOutboundBuffer 是否发生变化,只需要考虑【第 30 行】代码的判断。如果加了
!first
的判断,导致的结果是在 WriterIdleTimeoutTask 和 AllIdleTimeoutTask 任务中,ChannelOutboundBuffer 即使发生了变化,在首次还是会触发 write 和 all 空闲事件,在非首次不会触发 write 和 all 空闲事件。 - 关于上述的困惑,《Netty 那些事儿 ——— 关于 “Netty 发送大数据包时 触发写空闲超时” 的一些思考》 一文的作者,也表达了相同的困惑。后续,找闪电侠面基沟通下。
- 关于上述的困惑,《Netty 心跳服务之 IdleStateHandler 源码分析》 一文的作者,表达了自己的理解。感兴趣的胖友,可以看看。
- 当然,这块如果不理解的胖友,也不要方。从笔者目前了解下来,
observeOutput
都是设置为false
。也就说,不会触发这个方法的执行。
- 这是一个有点“神奇”的设定,笔者表示不太理解。理论来说,ChannelOutboundBuffer 是否发生变化,只需要考虑【第 30 行】代码的判断。如果加了
- 第 42 行:返回
false
,表示 ChannelOutboundBuffer 未发生变化。
5.8 AbstractIdleTask
AbstractIdleTask ,实现 Runnable 接口,空闲任务抽象类。代码如下:
AbstractIdleTask 是 IdleStateHandler 的内部静态类。
private abstract static class AbstractIdleTask implements Runnable { |
<1>
处,忽略未打开的 Channel 。<2>
处,子类实现#run()
抽象方法,实现自定义的空闲检测逻辑。
5.8.1 ReaderIdleTimeoutTask
ReaderIdleTimeoutTask ,继承 AbstractIdleTask 抽象类,检测 Read 空闲超时定时任务。代码如下:
ReaderIdleTimeoutTask 是 IdleStateHandler 的内部静态类。
1: private final class ReaderIdleTimeoutTask extends AbstractIdleTask { |
- 第 9 至 13 行:计算下一次检测的定时任务的延迟。
reading
为true
时,意味着正在读取,不会被检测为读空闲。reading
为false
时,实际nextDelay
的计算为readerIdleTimeNanos - (ticksInNanos() - lastReadTime)
。如果小于等于 0 ,意味着ticksInNanos() - lastReadTime >= readerIdleTimeNanos
,超时。
- ① 第 35 至 40 行:如果大于 0 ,说明未检测到读空闲。
- 第 39 行:调用
#schedule(ChannelHandlerContext ctx, Runnable task, long delay, TimeUnit unit)
方法,初始下一次的 ReaderIdleTimeoutTask 定时任务。其中,延迟时间为nextDelay
,即按照最后一次读的时间作为开始计数。
- 第 39 行:调用
- ② 第 15 至 34 行:如果小于等于 0 ,说明检测到读空闲。
- 第 19 行:调用
#schedule(ChannelHandlerContext ctx, Runnable task, long delay, TimeUnit unit)
方法,初始下一次的 ReaderIdleTimeoutTask 定时任务。其中,延迟时间为readerIdleTimeNanos
,即重新计数。 - 第 21 行:获得当前是否首次检测到读空闲。
- 第 24 行:标记
firstReaderIdleEvent
为false
。也就说,下次检测到空闲,就非首次了。
- 第 24 行:标记
- 第 28 行:调用
#newIdleStateEvent(IdleState state, boolean first)
方法,创建创建读空闲事件。- 第 30 行: 调用
#channelIdle(ChannelHandlerContext ctx, IdleStateEvent evt)
方法,在 pipeline 中,触发 UserEvent 事件。
- 第 30 行: 调用
- 第 31 至 34 行:如果发生异常,触发 Exception Caught 事件到下一个节点,处理异常。
- 第 19 行:调用
5.8.2 WriterIdleTimeoutTask
WriterIdleTimeoutTask ,继承 AbstractIdleTask 抽象类,检测 Write 空闲超时定时任务。代码如下:
WriterIdleTimeoutTask 是 IdleStateHandler 的内部静态类。
1: private final class WriterIdleTimeoutTask extends AbstractIdleTask { |
- 第 9 至 11 行:计算下一次检测的定时任务的延迟。
- ① 第 38 至 42 行:如果大于 0 ,说明未检测到写空闲。
- 第 39 行:调用
#schedule(ChannelHandlerContext ctx, Runnable task, long delay, TimeUnit unit)
方法,初始下一次的 WriterIdleTimeoutTask 定时任务。其中,延迟时间为nextDelay
,即按照最后一次写的时间作为开始计数。
- 第 39 行:调用
- ② 第 13 至 37 行:如果小于等于 0 ,说明检测到写空闲。
- 第 17 行:调用
#schedule(ChannelHandlerContext ctx, Runnable task, long delay, TimeUnit unit)
方法,初始下一次的 WriterIdleTimeoutTask 定时任务。其中,延迟时间为readerIdleTimeNanos
,即重新计数。 - 第 20 行:获得当前是否首次检测到写空闲。
- 第 22 行:标记
firstWriterIdleEvent
为false
。也就说,下次检测到空闲,就非首次了。
- 第 22 行:标记
- 第 25 至 28 行:判断 ChannelOutboundBuffer 是否发生变化。如果有变化,不触发写空闲时间。
- 第 31 行:调用
#newIdleStateEvent(IdleState state, boolean first)
方法,创建创建写空闲事件。- 第 33 行: 调用
#channelIdle(ChannelHandlerContext ctx, IdleStateEvent evt)
方法,在 pipeline 中,触发 UserEvent 事件。
- 第 33 行: 调用
- 第 34 至 37 行:如果发生异常,触发 Exception Caught 事件到下一个节点,处理异常。
- 第 17 行:调用
5.8.3 AllIdleTimeoutTask
AllIdleTimeoutTask ,继承 AbstractIdleTask 抽象类,检测 All 空闲超时定时任务。代码如下:
AllIdleTimeoutTask 是 IdleStateHandler 的内部静态类。
private final class AllIdleTimeoutTask extends AbstractIdleTask { |
- 因为 All 是 Write 和 Read 任一一种空闲即可,所以 AllIdleTimeoutTask 是 ReaderIdleTimeoutTask 和 WriterIdleTimeoutTask 的综合。
<1>
处,取lastReadTime
和lastWriteTime
中的大值,从而来判断,是否有 Write 和 Read 任一一种空闲。- WriterIdleTimeoutTask 就不详细解析,胖友自己读读代码即可。
6. ReadTimeoutHandler
io.netty.handler.timeout.ReadTimeoutHandler
,继承 IdleStateHandler 类,当 Channel 的读空闲时间( 读或者写 )太长时,抛出 ReadTimeoutException 异常,并自动关闭该 Channel 。
6.1 构造方法
/** |
closed
属性,Channel 是否关闭。<1>
处,禁用 Write / All 的空闲检测,只根据timeout
方法参数,开启 Read 的空闲检测。
6.2 channelIdle
#channelIdle(ChannelHandlerContext ctx, IdleStateEvent evt)
方法,覆写父类方法,代码如下:
|
<1>
处,触发 Exception Caught 事件到 pipeline 中,异常为 ReadTimeoutException 。<2>
处,关闭 Channel 通道。<3>
处,标记 Channel 为已关闭。
7. WriteTimeoutHandler
io.netty.handler.timeout.WriteTimeoutHandler
,继承 ChannelOutboundHandlerAdapter 类,当一个写操作不能在指定时间内完成时,抛出 WriteTimeoutException 异常,并自动关闭对应 Channel 。
😈 注意,这里写入,指的是 flush 到对端 Channel ,而不仅仅是写到 ChannelOutboundBuffer 队列。
7.1 构造方法
/** |
timeoutNanos
属性,写入超时时间,单位:纳秒。MIN_TIMEOUT_NANOS
属性,最小的超时时间,单位:纳秒。
lastTask
属性,WriteTimeoutTask 双向链表。其中,lastTask
为链表的尾节点。closed
属性,Channel 是否关闭。
7.2 handlerRemoved
#handlerRemoved(ChannelHandlerContext ctx)
方法,移除所有 WriteTimeoutTask 任务,并取消。代码如下:
|
- 代码比较简单,胖友自己看注释。
7.3 write
#write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
方法,代码如下:
|
<1>
处,如果promise
类型是 VoidPromise ,则包装成非 VoidPromise ,为了后续的回调。因为 VoidPromise 无法接收到回调。<2>
处,调用#scheduleTimeout(final ChannelHandlerContext ctx, final ChannelPromise promise)
方法,创建定时任务。详细解析,见 「7.4 scheduleTimeout」 。
7.4 scheduleTimeout
#scheduleTimeout(final ChannelHandlerContext ctx, final ChannelPromise promise)
方法,创建定时任务。代码如下:
1: private void scheduleTimeout(final ChannelHandlerContext ctx, final ChannelPromise promise) { |
- 第 2 至 6 行:创建 WriteTimeoutTask 任务,并发起定时任务。
- 第 8 行:如果定时任务已经执行完成,则不需要进行监听。否则,需要执行【第 10 至 14 行】的代码逻辑。
- 第 10 行:调用
#addWriteTimeoutTask(WriteTimeoutTask task)
方法,添加到链表。详细解析,见 「7.5 addWriteTimeoutTask」 。 - 第 14 行:将
task
作为监听器,添加到promise
中。在写入完成后,可以移除该定时任务。也就说,调用链是flush => 回调 => promise => 回调 => task
。
7.5 addWriteTimeoutTask
#addWriteTimeoutTask(WriteTimeoutTask task)
方法,添加到链表。代码如下:
private void addWriteTimeoutTask(WriteTimeoutTask task) { |
添加到链表的尾节点,并修改 lastTask
为当前任务。
7.6 removeWriteTimeoutTask
#removeWriteTimeoutTask(WriteTimeoutTask task)
方法,移除出链表。代码如下:
private void removeWriteTimeoutTask(WriteTimeoutTask task) { |
该方法的调用,在 「7.8 WriteTimeoutTask」 会看到。
7.7 writeTimedOut
#writeTimedOut(ChannelHandlerContext ctx)
方法,写入超时,关闭 Channel 通道。代码如下:
/** |
- 和
ReadTimeoutHandler#readTimeout(ChannelHandlerContext ctx)
方法,基本类似。
该方法的调用,在 「7.8 WriteTimeoutTask」 会看到。
7.8 WriteTimeoutTask
WriteTimeoutTask ,实现 Runnable 和 ChannelFutureListener 接口,写入超时任务。
WriteTimeoutTask 是 WriteTimeoutHandler 的内部类。
7.8.1 构造方法
private final ChannelHandlerContext ctx; |
7.8.2 run
当定时任务执行,说明写入任务执行超时。代码如下:
|
<1>
处,调用#writeTimedOut(ChannelHandlerContext ctx)
方法,写入超时,关闭 Channel 通道。<2>
处,调用#removeWriteTimeoutTask(WriteTimeoutTask task)
方法,移除出链表。
7.8.3 operationComplete
当回调方法执行,说明写入任务执行完成。代码如下:
|
<1>
处,取消定时任务。<2>
处,调用#removeWriteTimeoutTask(WriteTimeoutTask task)
方法,移除出链表。
666. 彩蛋
和 「5.7 hasOutputChanged」(#) 小节,这个方法较真了好久。感谢中间,基友【莫那一鲁道】的沟通。
推荐阅读文章:
- 莫那一鲁道 《Netty 心跳服务之 IdleStateHandler 源码分析》
- Hypercube 自顶向下深入分析Netty(八)–ChannelHandler