Netty空闲检测之写空闲
在之前的文章,我们介绍了Netty空闲检测之读空闲,以及为了介绍此篇文章,我们也特意写了一篇关于写操作的概括文章.读者对于Netty如何进行写操作也有了一个大概的认识了,接下来我们说一下,对于如何检测写空闲,Netty是如何控制的?
我们在向Pipeline中添加Handler的时候,绝大多数都会添加如下几个Handler.
分别是编码器(把写入外部地数据进行编码),解码器(把从外部读取地数据进行解码),空闲检测(检测是否读/写空闲),连接管理(如果存在空闲连接,如何处理),业务处理器(处理业务)
假如网络中发送过来一些数据,当这些数据被Netty读取,经过解码器解码之后,空闲检测中的读空闲拿到的数据,一定是一个完整的包含业务含义的数据(对象).然而,写操作就不是这样'完整'了.业务处理器向Netty写入一个数据(对象)之后,Netty可能只是暂时把一部分数据写入了Netty的缓冲区,另一部分写到了TCP缓冲区.
如上图所示,业务线程向Netty写入一个'HelloWorld'数据,可是会存在,Hello被写入到了TCP缓冲区,而World还在Netty的缓冲区中.当然这种情况一定会发生,但不是一直发生.
在线程池的知识里,有Future对象,我们可以调用get()方法获取任务的结果,但是这样会阻塞调用get()方法的线程(假如任务还没有完成).而Netty优化了这一方面,使得Netty是异步执行任务的.
要想实现和使用Netty的异步任务执行,必须按照上面的逻辑图码代码. 将业务线程抽离出来,只负责业务上的逻辑,具体的写操作由IO线程来完成,而且业务线程和IO线程都有一个队列,业务线程向IO线程的队列中放入写任务就返回,IO线程完成写之后,向业务线程的队列中放入一个通知任务,业务线程会及时感知到这个通知任务,然后调用业务线程之前设置的监听回调方法.我们直接看下IdleStateHandler类的源码是如何体现的.
// 源码位置: io.netty.handler.timeout.IdleStateHandler#write
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
if (writerIdleTimeNanos > 0 || allIdleTimeNanos > 0) {
// 业务线程在执行写入的时候,最终会将写操作封装成一个写任务放入IO线程的队列中.同时它会设置一个监听,用于回调使用.
ctx.write(msg, promise.unvoid()).addListener(writeListener);
} else {
ctx.write(msg, promise);
}
}
private final ChannelFutureListener writeListener = new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
lastWriteTime = ticksInNanos();
firstWriterIdleEvent = firstAllIdleEvent = true;
}
};以上说了这些都是和写操作有关,也和空闲检测之写空闲有关.下面我们来分析写空闲如何控制的.
A点是我们上次写空闲的检测时间点,B点是我们最后一次写操作的时间点,假如此时触发了写空闲检测,时间点在C点.
而B到C之间的时间长度并没有超过一个空闲检测的时间步长L(正如读空闲有个空闲检测的时间步长一样,写空闲也有一个空闲检测的时间步长),因为A到C之间才是一个时间步长L,因此空闲检测需要继续等待,但是,下一次的空闲检测不能是长度L了,而是A到B的时间长度,相当于在B到C这个时间段我已经检测了,现在只是不足一个时间步长L,我顶多再补偿剩下的时间就可以,因此下次的空闲检测时间步长是(B-A)的长度.
假如这个时候,空闲检测到了D点,还是没有发生写操作完成.此时B到D之间已经是一个完整的时间步长L了.但是呢,我们并不能说此时没有写操作.上面我们也说了,Netty有可能在缓慢地将数据从Netty的缓冲区写入到TCP的缓冲区,因为只有一个完整的数据写完,才能执行回调,更新最新的写操作时间.接下来空闲检测就会按照步长L的长度,再进行检测.至于我们是否要考虑刚才这种情况,Netty也是在我们构造IdleStateHandler类的时候可以传入一个observeOutput属性用于是否开启.默认是关闭的.
// 源码位置: io.netty.handler.timeout.IdleStateHandler.WriterIdleTimeoutTask
private final class WriterIdleTimeoutTask extends AbstractIdleTask {
WriterIdleTimeoutTask(ChannelHandlerContext ctx) {
super(ctx);
}
@Override
protected void run(ChannelHandlerContext ctx) {
long lastWriteTime = IdleStateHandler.this.lastWriteTime;
long nextDelay = writerIdleTimeNanos - (ticksInNanos() - lastWriteTime);
if (nextDelay <= 0) {
writerIdleTimeout = schedule(ctx, this, writerIdleTimeNanos, TimeUnit.NANOSECONDS);
boolean first = firstWriterIdleEvent;
firstWriterIdleEvent = false;
try {
// 这个地方就是用于是否考虑写操作慢的情况
if (hasOutputChanged(ctx, first)) {
return;
}
IdleStateEvent event = newIdleStateEvent(IdleState.WRITER_IDLE, first);
channelIdle(ctx, event);
} catch (Throwable t) {
ctx.fireExceptionCaught(t);
}
} else {
writerIdleTimeout = schedule(ctx, this, nextDelay, TimeUnit.NANOSECONDS);
}
}
}腾讯云开发者
