devbox
很楠不爱3
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Netty源码解析-IdleStateHandler

作者：很楠不爱3 | 2024-03-15 15:13:32

赞

踩

idlestatehandler

前言：

心跳机制广泛运用在我们的应用平台中。对于连接到应用服务的客户端，服务端有必要对长时间没有请求的客户端连接进行清理，以避免连接过多。这就需要服务端有空闲连接检测机制。

而针对客户端而言，如果长时间未请求数据，为避免被服务端清理连接，就需要间歇性的发送心跳请求。

在Netty中，针对以上需求，已经有现成的Handler可供使用，这就是本文要介绍的IdleStateHandler。

1.IdleStateHandler的使用

1.1 服务端检测长时间未请求的客户端

我们在服务端可以使用IdleStateHandler来检测长时间未发送请求的客户端，对其进行清理操作，简单示例如下：

  1. // 还是使用HelloServer的示例，我们在ChannelInitializer中添加IdleStateHandler
  2. .childHandler(new ChannelInitializer<SocketChannel>() {
  3.   protected void initChannel(SocketChannel ch) throws Exception {
  4.     ChannelPipeline pipeline = ch.pipeline();
  5.     // 在这里添加IdleStateHandler，设置空闲检测时间为10秒
  6.     pipeline.addLast("idle", new IdleStateHandler(10, 10, 10));
  7.     // 针对空闲事件的处理（自定义），具体内容如下
  8.     pipeline.addLast("idledeal", new IdleEventHandler());
  9.     pipeline.addLast("decoder", new StringDecoder());
  10.     pipeline.addLast("encoder", new StringEncoder());
  11.     pipeline.addLast("handler", new HelloServerHandler());
  12. }
  13. });
  14.  
  15. // IdleEventHandler
  16. public class IdleEventHandler extends ChannelDuplexHandler {
  17.  
  18.     @Override
  19.     public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
  20.         if (evt instanceof IdleStateEvent) {
  21.             evt = (IdleStateEvent) evt;
  22.  
  23.             // 若检测到长时间未读到请求事件，则清理客户端连接
  24.             if (evt.equals(IdleStateEvent.READER_IDLE_STATE_EVENT)) {
  25.                 System.out.println("idle...");
  26.                 ctx.channel().close();
  27.             }else {
  28.                 // 其他事件
  29.                 // TODO
  30.             }
  31.         }
  32.     }
  33. }

在本例中，我们设置了IdleStateHandler的读空闲检测时间为10s，则客户端连接10s没有发送任何请求过来时，则发送一个IdleStateEvent.READER_IDLE_STATE_EVENT事件到下游，IdleEventHandler处理该事件，直接关闭客户端连接。

1.2 客户端发送心跳请求

若客户端本身检测到长时间未发送请求，为避免被服务端清理，则可以主动发送一个心跳请求。简单示例如下

  1. // 同样的使用HelloClient的代码，我们改造下ChannelInitializer
  2. .handler(new ChannelInitializer<SocketChannel>() {
  3.   protected void initChannel(SocketChannel ch) throws Exception {
  4.     ChannelPipeline pipeline = ch.pipeline();
  5.     ...
  6.     // 在这里添加IdleStateHandler检测
  7.     pipeline.addLast("idle", new IdleStateHandler(5, 5, 5));
  8.     // 空闲事件处理Handler（自定义），具体内容如下
  9.     pipeline.addLast("idledeal", new ClientIdleEventHandler());
  10.     pipeline.addLast("handler", new HelloClientHandler());
  11. }
  12. });
  13.  
  14. // ClientIdleEventHandler
  15. public class ClientIdleEventHandler extends ChannelDuplexHandler {
  16.     @Override
  17.     public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
  18.         if (evt instanceof IdleStateEvent) {
  19.             evt = (IdleStateEvent) evt;
  20.  
  21.             // 若检测到长时间未发送请求事件，则主动发送心跳信息
  22.             if (evt.equals(IdleStateEvent.WRITER_IDLE_STATE_EVENT)) {
  23.                 ctx.writeAndFlush("ping");
  24.             }else {
  25.                 // 其他事件
  26.                 // TODO
  27.             }
  28.         }
  29.     }
  30. }

这样，当客户端发现已经5s没有发送过请求时，则主动发送一个ping心跳信息到服务端，避免被清理

2.IdleStateHandler的构造

我们首先来看下IdleStateHandler的相关构造方法和基本属性

  1. public class IdleStateHandler extends ChannelDuplexHandler {
  2.     
  3.     // Not create a new ChannelFutureListener per write operation to reduce GC pressure.
  4.     // write监听器
  5.     private final ChannelFutureListener writeListener = new ChannelFutureListener() {
  6.         @Override
  7.         public void operationComplete(ChannelFuture future) throws Exception {
  8.             lastWriteTime = ticksInNanos();
  9.             firstWriterIdleEvent = firstAllIdleEvent = true;
  10.         }
  11.     };
  12.  
  13.     private final boolean observeOutput;
  14. 	// 三种类型的空闲时间设置
  15.     private final long readerIdleTimeNanos;
  16.     private final long writerIdleTimeNanos;
  17.     private final long allIdleTimeNanos;
  18.  
  19. 	// 读空闲检测定时任务
  20.     private ScheduledFuture<?> readerIdleTimeout;
  21. 	// 最近一次读事件
  22.     private long lastReadTime;
  23. 	// 是否第一次读idleEvent触发
  24.     private boolean firstReaderIdleEvent = true;
  25.  
  26. 	// 以下与读设置类似
  27.     private ScheduledFuture<?> writerIdleTimeout;
  28.     private long lastWriteTime;
  29.     private boolean firstWriterIdleEvent = true;
  30.  
  31.     private ScheduledFuture<?> allIdleTimeout;
  32.     private boolean firstAllIdleEvent = true;
  33.  
  34. 	// IdleStateHandler的状态，避免多次初始化
  35.     private byte state; // 0 - none, 1 - initialized, 2 - destroyed
  36.     private boolean reading;
  37.  
  38.     private long lastChangeCheckTimeStamp;
  39.     private int lastMessageHashCode;
  40.     private long lastPendingWriteBytes;
  41.     private long lastFlushProgress;
  42.  
  43. 	// 默认使用的构造器
  44. 	public IdleStateHandler(
  45.             int readerIdleTimeSeconds,
  46.             int writerIdleTimeSeconds,
  47.             int allIdleTimeSeconds) {
  48.  
  49.         // 默认单位为秒
  50.         this(readerIdleTimeSeconds, writerIdleTimeSeconds, allIdleTimeSeconds,
  51.              TimeUnit.SECONDS);
  52.     }
  53.  
  54.     public IdleStateHandler(
  55.             long readerIdleTime, long writerIdleTime, long allIdleTime,
  56.             TimeUnit unit) {
  57.         this(false, readerIdleTime, writerIdleTime, allIdleTime, unit);
  58.     }
  59.  
  60. 	public IdleStateHandler(boolean observeOutput,
  61.             long readerIdleTime, long writerIdleTime, long allIdleTime,
  62.             TimeUnit unit) {
  63.         ObjectUtil.checkNotNull(unit, "unit");
  64.  
  65.         this.observeOutput = observeOutput;
  66.  
  67.         // 以纳秒为单位重新设置超时时间
  68.         if (readerIdleTime <= 0) {
  69.             readerIdleTimeNanos = 0;
  70.         } else {
  71.             readerIdleTimeNanos = Math.max(unit.toNanos(readerIdleTime), MIN_TIMEOUT_NANOS);
  72.         }
  73.         if (writerIdleTime <= 0) {
  74.             writerIdleTimeNanos = 0;
  75.         } else {
  76.             writerIdleTimeNanos = Math.max(unit.toNanos(writerIdleTime), MIN_TIMEOUT_NANOS);
  77.         }
  78.         if (allIdleTime <= 0) {
  79.             allIdleTimeNanos = 0;
  80.         } else {
  81.             allIdleTimeNanos = Math.max(unit.toNanos(allIdleTime), MIN_TIMEOUT_NANOS);
  82.         }
  83.     }
  84. }

IdleStateHandler继承了ChannelDuplexHandler，说明其可以处理inbound、outbound事件；

构造方法比较简单，我们比较常用的就是第一个构造方法，以秒为单位来设置读、写、all的空闲检测；

属性的话，我们通过具体方法来学习。

3.IdleStateHandler空闲检测

在handlerAdded、channelActive等方法中，都有一个initialize()方法，这个方法用来初始化检测器，我们先来看下

3.1 initialize() 初始化IdleStateHandler

  1. public class IdleStateHandler extends ChannelDuplexHandler {
  2.     private void initialize(ChannelHandlerContext ctx) {
  3.         // 已经初始化过则不再重复初始化
  4.         switch (state) {
  5.             case 1:
  6.             case 2:
  7.                 return;
  8.         }
  9.  
  10.         // 设置状态为initialized
  11.         state = 1;
  12.         // TODO
  13.         initOutputChanged(ctx);
  14.  
  15.         // 设置最新的readTime和writeTime为当前时间
  16.         lastReadTime = lastWriteTime = ticksInNanos();
  17.         if (readerIdleTimeNanos > 0) {
  18.             // 创建一个读空闲检测定时任务，延后readerIdleTimeNanos执行，具体schedule方法见下面
  19.             readerIdleTimeout = schedule(ctx, new ReaderIdleTimeoutTask(ctx),
  20.                                          readerIdleTimeNanos, TimeUnit.NANOSECONDS);
  21.         }
  22.         if (writerIdleTimeNanos > 0) {
  23.             // 创建一个写空闲检测定时任务，延后readerIdleTimeNanos执行
  24.             writerIdleTimeout = schedule(ctx, new WriterIdleTimeoutTask(ctx),
  25.                                          writerIdleTimeNanos, TimeUnit.NANOSECONDS);
  26.         }
  27.         if (allIdleTimeNanos > 0) {
  28.             // 创建一个读、写空闲检测定时任务，延后readerIdleTimeNanos执行
  29.             allIdleTimeout = schedule(ctx, new AllIdleTimeoutTask(ctx),
  30.                                       allIdleTimeNanos, TimeUnit.NANOSECONDS);
  31.         }
  32.     }
  33.     
  34.     //  创建一个定时任务
  35.     ScheduledFuture<?> schedule(ChannelHandlerContext ctx, Runnable task, long delay, TimeUnit unit) {
  36.         return ctx.executor().schedule(task, delay, unit);
  37.     }
  38. }

初始化方法，主要用于初始化三个定时任务，那么这三个定时任务ReaderIdleTimeoutTask、WriterIdleTimeoutTask、AllIdleTimeoutTask具体是怎么用的呢？我们先来看下读空闲检测是如何做的。

3.2 读空闲检测

  1. public class IdleStateHandler extends ChannelDuplexHandler {
  2. 	@Override
  3.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  4.         // 当设置的readerIdleTimeNanos或allIdleTimeNanos大于0时，说明需要进行读空闲检测
  5.         if (readerIdleTimeNanos > 0 || allIdleTimeNanos > 0) {
  6.             // 设置reading 正在读数据的状态为true
  7.             reading = true;
  8.             // 设置两个状态为为true
  9.             firstReaderIdleEvent = firstAllIdleEvent = true;
  10.         }
  11.         ctx.fireChannelRead(msg);
  12.     }
  13.  
  14.     // 重点在这来，如果本次读已经结束，则需要重置时间和状态位
  15.     @Override
  16.     public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
  17.         // 如果数据正在读状态
  18.         if ((readerIdleTimeNanos > 0 || allIdleTimeNanos > 0) && reading) {
  19.             // 则重新设置lastReadTime为当前时间
  20.             lastReadTime = ticksInNanos();
  21.             // 将正在读标志设置为false
  22.             reading = false;
  23.         }
  24.         ctx.fireChannelReadComplete();
  25.     }
  26. }

通过上述两个方法可以看出，当发生读事件时，设置reading=true，当本次读结束时，则设置reading=false，lastReadTime（最近一次读时间）为当前时间。

那么这个是如何被检测到读超时的呢？我们可以回到initialize()方法，其中有一个ReaderIdleTimeoutTask的定时任务，延迟readerIdleTimeNanos执行，一起来看下这个task的具体内容

3.2.1 ReaderIdleTimeoutTask 读空闲检测定时任务

  1. private final class ReaderIdleTimeoutTask extends AbstractIdleTask {
  2.  
  3.     // 将ChannelHandlerContext传入当前task
  4.     ReaderIdleTimeoutTask(ChannelHandlerContext ctx) {
  5.         super(ctx);
  6.     }
  7.  
  8.     @Override
  9.     protected void run(ChannelHandlerContext ctx) {
  10.         long nextDelay = readerIdleTimeNanos;
  11.         // 如果当前没有发生读事件，则reading为false
  12.         if (!reading) {
  13.             nextDelay -= ticksInNanos() - lastReadTime;
  14.         }
  15.  
  16.         // 读空闲超时，需要发送READER_IDLE事件
  17.         if (nextDelay <= 0) {
  18.             // Reader is idle - set a new timeout and notify the callback.
  19.             // 重启一个定时检测任务，延迟readerIdleTimeNanos执行
  20.             readerIdleTimeout = schedule(ctx, this, readerIdleTimeNanos, TimeUnit.NANOSECONDS);
  21.  
  22.             boolean first = firstReaderIdleEvent;
  23.             firstReaderIdleEvent = false;
  24.  
  25.             try {
  26.                 // 往下游发送一个READER_IDLE Event
  27.                 IdleStateEvent event = newIdleStateEvent(IdleState.READER_IDLE, first);
  28.                 channelIdle(ctx, event);
  29.             } catch (Throwable t) {
  30.                 ctx.fireExceptionCaught(t);
  31.             }
  32.         } else {
  33.             // 注意这里虽然也是重启一个定时任务，但是延迟时间与上面有所不同，这里的具体延迟时间为readerIdleTimeNanos - (ticksInNanos() - lastReadTime)
  34.             readerIdleTimeout = schedule(ctx, this, nextDelay, TimeUnit.NANOSECONDS);
  35.         }
  36.     }
  37.     
  38.     protected void channelIdle(ChannelHandlerContext ctx, IdleStateEvent evt) throws Exception {
  39.         ctx.fireUserEventTriggered(evt);
  40.     }
  41. }

这里比较有意思的是nextDelay值的设置，当读事件正在进行时（reading=true），则直接进行下一次循环；

当读事件未执行，若ticksInNanos（当前时间） - lastReadTime（最后一次读完成时间） > readerIdleTimeNanos（读空闲检测时间），说明读空闲超时，往下游发送一个READER_IDLE Event

reading=false的情况有两种：没有发生过读、读已经结束；当读数据正在进行时，则reading=true

总结：通过这种对lastReadTime的定时任务检测，就可以发现是否已经长时间未读，若是，则发送下游READER_IDLE事件，下游检测到该事件进行相应处理即可。

3.3 写空闲检测

分析过程与3.2 读空闲检测类似，我们先来看下重写后的write方法

  1. public class IdleStateHandler extends ChannelDuplexHandler {
  2. 	@Override
  3.     public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
  4.         // 如果writerIdleTimeNanos或allIdleTimeNanos大于0，说明需要进行写空闲检测
  5.         if (writerIdleTimeNanos > 0 || allIdleTimeNanos > 0) {
  6.             // 对write方法执行后的ChannelFuture添加监听器，writeListener内容如下
  7.             ctx.write(msg, promise.unvoid()).addListener(writeListener);
  8.         } else {
  9.             ctx.write(msg, promise);
  10.         }
  11.     }
  12.     
  13.     private final ChannelFutureListener writeListener = new ChannelFutureListener() {
  14.         @Override
  15.         public void operationComplete(ChannelFuture future) throws Exception {
  16.             // 当写方法完成时，设置最新一次写时间为当前时间
  17.             lastWriteTime = ticksInNanos();
  18.             firstWriterIdleEvent = firstAllIdleEvent = true;
  19.         }
  20.     };
  21. }

方法并不复杂，主要就是对write方法添加一个监听器，用于监听wirte方法完成，完成后重置下lastWriteTime。下面来看下WriteTask所做的事情

3.3.1 WriterIdleTimeoutTask 写空闲检测

  1. private final class WriterIdleTimeoutTask extends AbstractIdleTask {
  2.  
  3.     // 将ChannelHandlerContext传入当前task
  4.     WriterIdleTimeoutTask(ChannelHandlerContext ctx) {
  5.         super(ctx);
  6.     }
  7.  
  8.     @Override
  9.     protected void run(ChannelHandlerContext ctx) {
  10.         long lastWriteTime = IdleStateHandler.this.lastWriteTime;
  11.         // 同样的方式来计算nextDelay
  12.         long nextDelay = writerIdleTimeNanos - (ticksInNanos() - lastWriteTime);
  13.         // 已超时
  14.         if (nextDelay <= 0) {
  15.             // 先生成一个定时任务，用于下次写超时检测
  16.             writerIdleTimeout = schedule(ctx, this, writerIdleTimeNanos, TimeUnit.NANOSECONDS);
  17.  
  18.             boolean first = firstWriterIdleEvent;
  19.             firstWriterIdleEvent = false;
  20.  
  21.             try {
  22.                 // 这里比较有意思，我们具体在3.3.2 来看下
  23.                 if (hasOutputChanged(ctx, first)) {
  24.                     return;
  25.                 }
  26.  
  27.                 // 直接向下游传递一个WRITER_IDLE事件
  28.                 IdleStateEvent event = newIdleStateEvent(IdleState.WRITER_IDLE, first);
  29.                 channelIdle(ctx, event);
  30.             } catch (Throwable t) {
  31.                 ctx.fireExceptionCaught(t);
  32.             }
  33.         } else {
  34.             // 说明写未超时，重新生成一个定时任务，延迟nextDelay执行
  35.             writerIdleTimeout = schedule(ctx, this, nextDelay, TimeUnit.NANOSECONDS);
  36.         }
  37.     }
  38. }

3.3.2 hasOutputChanged() 判断Channeloutboundbuffer是否发生过变化

  1. public class IdleStateHandler extends ChannelDuplexHandler {
  2. 	private boolean hasOutputChanged(ChannelHandlerContext ctx, boolean first) {
  3.         // 默认observeOutput=false，不会进行下面的检测，需要主动开启
  4.         if (observeOutput) {
  5.  
  6.             // 如果 lastChangeCheckTimeStamp 和 lastWriteTime 不一样，说明写操作进行过了，需要更新此值
  7.             if (lastChangeCheckTimeStamp != lastWriteTime) {
  8.                 lastChangeCheckTimeStamp = lastWriteTime;
  9.  
  10.                 // 非首次，则直接返回true
  11.                 // 这里的first，即firstWriterIdleEvent参数，默认为true，当写操作完成时也被置为true
  12.                 if (!first) {
  13.                     return true;
  14.                 }
  15.             }
  16.  
  17.             Channel channel = ctx.channel();
  18.             Unsafe unsafe = channel.unsafe();
  19.             ChannelOutboundBuffer buf = unsafe.outboundBuffer();
  20.  
  21.             if (buf != null) {
  22.                 int messageHashCode = System.identityHashCode(buf.current());
  23.                 long pendingWriteBytes = buf.totalPendingWriteBytes();
  24.  
  25.                 // 这来主要判断ChannelOutboundBuffer中的值是否发生了变化
  26.                 // 如果有数据添加进来，则前后肯定不一致
  27.                 if (messageHashCode != lastMessageHashCode || pendingWriteBytes != lastPendingWriteBytes) {
  28.                     lastMessageHashCode = messageHashCode;
  29.                     lastPendingWriteBytes = pendingWriteBytes;
  30.  
  31.                     if (!first) {
  32.                         return true;
  33.                     }
  34.                 }
  35.  
  36.                 long flushProgress = buf.currentProgress();
  37.                 if (flushProgress != lastFlushProgress) {
  38.                     lastFlushProgress = flushProgress;
  39.  
  40.                     if (!first) {
  41.                         return true;
  42.                     }
  43.                 }
  44.             }
  45.         }
  46.  
  47.         return false;
  48.     }
  49. }

关于检测Channeloutboundbuffer变化的逻辑，在正常使用IdleStateHandler中是不会触发的，具体的细节分析可以参考下下文

Netty 心跳服务之 IdleStateHandler 源码分析 - 简书

总结：写空闲检测与读空闲检测基本类似，笔者不再赘述。

3.4 AllIdle检测

这个的检测与上述读写检测基本是一样的，大家可以自行阅读AllIdleTimeoutTask.java，笔者不再赘述

总结：

IdleStateHandler的代码不算复杂，在我们的应用探活中可以很好的发挥作用。

主要还是我们监听到Idle event后的自定义处理方案，这个才是关键。

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/很楠不爱3/article/detail/242191
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号