Netty 系列笔记之 keepalive 与 idle 监测

2023-03-18 / netty

Netty 系列笔记之 keepalive 与 idle 监测

一、开篇

在网络编程中，我们经常听到 keepalive 这个词。keepalive 是什么？为什么需要 keepalive ？idle 监测是什么？网络编程中到底怎么用？

本篇，我们就探讨 keepalive 与 idle 监测，以及 Netty 如何支持它们。

二、 keepalive

1、为什么需要 keepalive

以打电话为例，对方和你说这说着突然不讲话了。大多数情况下你不会你只等下去，你会和对方确认下：“你还在吗？”，如果对方没有回答，就挂掉电话。这套机制就是 keepalive 。

同样类比到服务器应用，服务器需要 keepalive 确保这条连接是正常的，才能够正常通信。

2、如何设计 keepalive ？（以 TCP 为例）

TCP keepalive 有三个核心参数：

net.ipv4.tcp_keepalive_time = 7200
net.ipv4.tcp_keepalive_intvl = 75
net.ipv4.tcp_keepalive_probes = 9

当启用 keepalive （默认关闭）时，TCP 连接 7200 秒后没有数据通过发送 keepalive 消息，当探测没有确认时，按照 75 秒的间隔发送重试，一共发送 9 次都没有收到确认，则认为此连接失效。

计算总耗时为 2 小时 11 分钟（7200 + 75 * 9）。

3、为什么应用层还需要 keepalive ？

在开发应用程序时，常遇到需要添加 “心跳” 的需求，既然 TCP 已经有 keepalive 了，为什么还需要应用层添加逻辑呢？原因如下：

协议分层，关注点不同。传输层关注连接是否通，而应用层关注服务是否可用。比如打电话，电话可以打通，但是不代表有人接听。服务器连接在，不一定提供服务。
TCP 的 keepalive 默认关闭，且经过路由器等中转设备时 keepalive 包可能被丢弃。
TCP 的 keepalive 时间太长，虽然可以改，但是不建议。一旦改动，会影响所有应用。

❤ 注意：在 HTTP 连接中，我们常常看到 keep-alive ，说的不是一回事。其代表长连接（keep-alive）和短链接（close）。

Content-Type: application/json; charset=utf-8
Transfer-Encoding: chunked
Connection: keep-alive

三、idle 监测

继续以打电话为例，对方说着说着没声音了，一般来说你会等待一段时间，在这个时间内看对方还说不说话（idle 监测），如果还不说话，就认为对方存在问题（idle），于是开始询问对方：“你还在吗？”（keepalive），或者直接挂掉电话（关闭连接）。

❀ idle 监测的功能

idle 监测只是负责诊断，诊断后作出不同的行为，决定 idle 监测的最终用途。

发送 keepalive ：一般来配合 keepalive ，来减少 keepalive 消息。
- keepalive 设计演进：
  v1 ：定时 keepalive 消息 -> v2 ：没有数据传输时进行空闲检测，判定为 idle 时才发送 keepalive 消息。
直接关闭连接：快速释放损坏的、很久不用的、恶意的连接，保持系统健康。

实际应用：结合起来使用，按需 keepalive ，保证不会空闲，如果空闲，关闭连接。

四、Netty 与 keepalive & idle

1、Netty 中开启 keepalive 和 idle

开启 keepalive
两种方式：
- bootstrap.childOption(ChannelOption.SO_KEEPALIVE, Boolean.TRUE);
- bootstrap.childOption(NioChannelOption.of(StandardSocketOptions.SO_KEEPALIVE), Boolean.TRUE);
开启 idle 监测
socketChannel.addLast("idleCheckHandler", new IdleStateHandler(0, 20, 0, TimeUnit.SECONDS));

如果参数为负数，则被设置为 0 ：

public IdleStateHandler(boolean observeOutput,
            long readerIdleTime, long writerIdleTime, long allIdleTime,
            TimeUnit unit) {
        if (unit == null) {
            throw new NullPointerException("unit");
        }

        this.observeOutput = observeOutput;

        if (readerIdleTime <= 0) {
            readerIdleTimeNanos = 0;
        } else {
            readerIdleTimeNanos = Math.max(unit.toNanos(readerIdleTime), MIN_TIMEOUT_NANOS);
        }
        if (writerIdleTime <= 0) {
            writerIdleTimeNanos = 0;
        } else {
            writerIdleTimeNanos = Math.max(unit.toNanos(writerIdleTime), MIN_TIMEOUT_NANOS);
        }
        if (allIdleTime <= 0) {
            allIdleTimeNanos = 0;
        } else {
            allIdleTimeNanos = Math.max(unit.toNanos(allIdleTime), MIN_TIMEOUT_NANOS);
        }
    }

2、源码分析

关于 keepalive 的设置比较简单，这里我们就不说了，主要看下 idle 的源码实现，关于 idle 的代码在 io.netty.handler.timeout 包下：

我们逐个来阅读：

(1) `IdleState`

public enum IdleState {
  /**
   * No data was received for a while.
   */
  READER_IDLE,
  /**
   * No data was sent for a while.
   */
  WRITER_IDLE,
  /**
   * No data was either received or sent for a while.
   */
  ALL_IDLE
}

idle 状态的枚举，包含三种情况

读空闲，一段时间内没有数据被接收。
写空闲，一段时间内没有数据被发送。
读或写任一空闲。

(2) `IdleStateEvent`

public class IdleStateEvent {
  public static final IdleStateEvent FIRST_READER_IDLE_STATE_EVENT =
          new DefaultIdleStateEvent(IdleState.READER_IDLE, true);
  public static final IdleStateEvent READER_IDLE_STATE_EVENT =
          new DefaultIdleStateEvent(IdleState.READER_IDLE, false);
  public static final IdleStateEvent FIRST_WRITER_IDLE_STATE_EVENT =
          new DefaultIdleStateEvent(IdleState.WRITER_IDLE, true);
  public static final IdleStateEvent WRITER_IDLE_STATE_EVENT =
          new DefaultIdleStateEvent(IdleState.WRITER_IDLE, false);
  public static final IdleStateEvent FIRST_ALL_IDLE_STATE_EVENT =
          new DefaultIdleStateEvent(IdleState.ALL_IDLE, true);
  public static final IdleStateEvent ALL_IDLE_STATE_EVENT =
          new DefaultIdleStateEvent(IdleState.ALL_IDLE, false);

  private final IdleState state;
  private final boolean first;

... 省略部分代码
}

这个类根据上面的三种状态定义了发生 idle 时触发的 6 个事件，为什么是 6 个不是 3 个呢？对于第一次发生 idle ，抽出了第一次发生的事件，这样就能只对第一次发生 idle 时做处理，后面就可以忽略。

(3) `IdleStateHandler`

IdleStateHandler extends ChannelDuplexHandler
继承自 ChannelDuplexHandler ，通过重写其 channelRead、write 等方法可以获得 channel 的读写状态并控制 IdleStateHandler 的生命周期。当读或写空闲时，将会触发 IdleStateEvent 事件。

构造方法

public IdleStateHandler(
        int readerIdleTimeSeconds,
        int writerIdleTimeSeconds,
        int allIdleTimeSeconds) {

    this(readerIdleTimeSeconds, writerIdleTimeSeconds, allIdleTimeSeconds,
         TimeUnit.SECONDS);
}

构造方法就是我们之前看到的设置开启 idle 的代码，其中三个参数分别对应了 idle 的三种状态。

属性

/**
   * 最小超时时间
   */
  private static final long MIN_TIMEOUT_NANOS = TimeUnit.MILLISECONDS.toNanos(1);
  /**
   * 写入监听器
   */
  // Not create a new ChannelFutureListener per write operation to reduce GC pressure.
  private final ChannelFutureListener writeListener = new ChannelFutureListener() {
      @Override
      public void operationComplete(ChannelFuture future) throws Exception {
          lastWriteTime = ticksInNanos();
          firstWriterIdleEvent = firstAllIdleEvent = true;
      }
  };
  /**
   * 是否观察 {@link ChannelOutboundBuffer} 写入队列
   */
  private final boolean observeOutput;
  /**
   * 读空闲时间
   */
  private final long readerIdleTimeNanos;
  /**
   * 写空闲时间
   */
  private final long writerIdleTimeNanos;
  /**
   * 读写任一空闲时间
   */
  private final long allIdleTimeNanos;
  /**
   * 读空闲的定时检测任务
   */
  private ScheduledFuture<?> readerIdleTimeout;
  /**
   * 最后读时间
   */
  private long lastReadTime;
  /**
   * 是否首次读空闲
   */
  private boolean firstReaderIdleEvent = true;
  /**
   * 写空闲的定时检测任务
   */
  private ScheduledFuture<?> writerIdleTimeout;
  /**
   * 最后写时间
   */
  private long lastWriteTime;
  /**
   * 是否第一次写空闲
   */
  private boolean firstWriterIdleEvent = true;
  /**
   * 读写任一空闲时间
   */
  private ScheduledFuture<?> allIdleTimeout;
  /**
   * 是否首次读写任一空闲
   */
  private boolean firstAllIdleEvent = true;
  /**
   * handler 状态
   * 0：未初始化
   * 1：初始化完毕
   * 2：销毁
   */
  private byte state; // 0 - none, 1 - initialized, 2 - destroyed
  /**
   * 是否正在读取
   */
  private boolean reading;
  /**
   * 最后检测到 {@link ChannelOutboundBuffer} 发生变化的时间
   */
  private long lastChangeCheckTimeStamp;
  /**
   * 上一次准备 flash 到对端的消息( {@link ChannelOutboundBuffer#current()} )的 HashCode
   */
  private int lastMessageHashCode;
  /**
   * 上一次等待 flush 到对端的内存大小( {@link ChannelOutboundBuffer#totalPendingWriteBytes()} )
   */
  private long lastPendingWriteBytes;
  /**
   * 上一次 flush 到对端的进度
   */
  private long lastFlushProgress;

observeOutput 这个属性是标识写空闲的判断依据是否是写成功。两种写不成功的情况：1）写了，但是缓冲区满了，写不出去；2）写了一个特别大的文件，是在写，但是还没写成功。上述两种情况确实是在写，但是不代表写空闲。所以可以调整这个参数按照实际需要决定写空闲的判断依据。

其中涉及到的一些不清楚的地方，我们在后面分析的时候会有提及。

initialize(ChannelHandlerContext ctx) 方法

private void initialize(ChannelHandlerContext ctx) {
      // Avoid the case where destroy() is called before scheduling timeouts.
      // See: https://github.com/netty/netty/issues/143
      // 根据状态判断是否执行，保证 destory() 方法在初始化后执行
      switch (state) {
          case 1:
          case 2:
              return;
      }
      // 设置状态为初始化完毕
      state = 1;
      // ChannelOutboundBuffer 相关属性初始化
      initOutputChanged(ctx);
      // 最后读写时间初始化
      lastReadTime = lastWriteTime = ticksInNanos();
      // 初始化三种事件的定时任务
      if (readerIdleTimeNanos > 0) {
          readerIdleTimeout = schedule(ctx, new ReaderIdleTimeoutTask(ctx),
                  readerIdleTimeNanos, TimeUnit.NANOSECONDS);
      }
      if (writerIdleTimeNanos > 0) {
          writerIdleTimeout = schedule(ctx, new WriterIdleTimeoutTask(ctx),
                  writerIdleTimeNanos, TimeUnit.NANOSECONDS);
      }
      if (allIdleTimeNanos > 0) {
          allIdleTimeout = schedule(ctx, new AllIdleTimeoutTask(ctx),
                  allIdleTimeNanos, TimeUnit.NANOSECONDS);
      }
  }

ReaderIdleTimeoutTask
我们知道判断读写空闲是使用定时任务的方式进行的，看下读空闲定时任务。

private final class ReaderIdleTimeoutTask extends AbstractIdleTask {

      ReaderIdleTimeoutTask(ChannelHandlerContext ctx) {
          super(ctx);
      }

      @Override
      protected void run(ChannelHandlerContext ctx) {
          // 下一次检测的定时任务的时间
          long nextDelay = readerIdleTimeNanos;
          // 不处于正在读取状态
          if (!reading) {
              // 配置的空闲时间 - 实际空闲时间
              nextDelay -= ticksInNanos() - lastReadTime;
          }
          // 小于 0 说明读空闲，也就是说空闲时间大于配置的读空闲时间
          if (nextDelay <= 0) {
              // 重新设置定时任务的空闲时间
              // Reader is idle - set a new timeout and notify the callback.
              readerIdleTimeout = schedule(ctx, this, readerIdleTimeNanos, TimeUnit.NANOSECONDS);
              // 标志是否是首次空闲
              boolean first = firstReaderIdleEvent;
              firstReaderIdleEvent = false;

              try {
                  // 创建读空闲事件
                  IdleStateEvent event = newIdleStateEvent(IdleState.READER_IDLE, first);
                  // 通知通道空闲事件
                  channelIdle(ctx, event);
              } catch (Throwable t) {
                  ctx.fireExceptionCaught(t);
              }
          } else {
              // 空闲时间设置为 nextDelay , 按照最后一次读的时间作为开始计数
              // Read occurred before the timeout - set a new timeout with shorter delay.
              readerIdleTimeout = schedule(ctx, this, nextDelay, TimeUnit.NANOSECONDS);
          }
      }
  }

WriterIdleTimeoutTask
检测写空闲的定时任务，注意写空闲如何定义，hasOutputChanged 方法用于判断写的过程是否发生变化，如果发生变化就认为此时没有发生写空闲而 return 。

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);
          // 小于 0 说明写空闲
          if (nextDelay <= 0) {
              // 重新设置定时任务的空闲时间
              // Writer is idle - set a new timeout and notify the callback.
              writerIdleTimeout = schedule(ctx, this, writerIdleTimeNanos, TimeUnit.NANOSECONDS);
              // 标志是否首次空闲
              boolean first = firstWriterIdleEvent;
              firstWriterIdleEvent = false;

              try {
                  // 判断 ChannelOutboundBuffer 是否发生变化
                  if (hasOutputChanged(ctx, first)) {
                      return;
                  }
                  // 创建写空闲事件
                  IdleStateEvent event = newIdleStateEvent(IdleState.WRITER_IDLE, first);
                  // 通知通道空闲事件
                  channelIdle(ctx, event);
              } catch (Throwable t) {
                  ctx.fireExceptionCaught(t);
              }
          } else {
              // Write occurred before the timeout - set a new timeout with shorter delay.
              writerIdleTimeout = schedule(ctx, this, nextDelay, TimeUnit.NANOSECONDS);
          }
      }
  }

hasOutputChanged 方法

其中的 observeOutput 属性在前面已经简单说过了，该方法用于检测 ChannelOutboundBuffer 队列是否发生变化，用来控制是否是写空闲的依据。

private boolean hasOutputChanged(ChannelHandlerContext ctx, boolean first) {
      if (observeOutput) {
          // 一般情况下，false , 即写空闲判断是否写成功。
          // 实际情况下：
          // 1) 写了，但是缓存区满了，写不出去；2）写了一个大数据，正在写，但是没有写完成。
          // 上述这两种情况，确实在写，但是还没有写成功。
          // 所以这个参数是判断 "是否在写" , 而不是 "是否写成功" 。
          // We can take this shortcut if the ChannelPromises that got passed into write()
          // appear to complete. It indicates "change" on message level and we simply assume
          // that there's change happening on byte level. If the user doesn't observe channel
          // writability events then they'll eventually OOME and there's clearly a different
          // problem and idleness is least of their concerns.
          // 如果 lastChangeCheckTimeStamp != lastWriteTime 说明进行过写操作，更新 lastChangeCheckTimeStamp
          if (lastChangeCheckTimeStamp != lastWriteTime) {
              lastChangeCheckTimeStamp = lastWriteTime;
              // 不是第一次
              // But this applies only if it's the non-first call.
              if (!first) {
                  return true;
              }
          }

          Channel channel = ctx.channel();
          Unsafe unsafe = channel.unsafe();
          ChannelOutboundBuffer buf = unsafe.outboundBuffer();

          if (buf != null) {
              int messageHashCode = System.identityHashCode(buf.current());
              long pendingWriteBytes = buf.totalPendingWriteBytes();
              // 此时的数据哈希与上一次的不同 || 正在等待发送的数据字节数不等于上次的字节数，说明发生了变化
              if (messageHashCode != lastMessageHashCode || pendingWriteBytes != lastPendingWriteBytes) {
                  lastMessageHashCode = messageHashCode;
                  lastPendingWriteBytes = pendingWriteBytes;

                  if (!first) {
                      return true;
                  }
              }
              // flush 的进度发生了变化
              long flushProgress = buf.currentProgress();
              if (flushProgress != lastFlushProgress) {
                  lastFlushProgress = flushProgress;

                  if (!first) {
                      return true;
                  }
              }
          }
      }

      return false;
  }

(4)`ReadTimeoutHandler`

前面说的 IdleStateHandler 作用是判断空闲，并创建了空闲事件，具体的处理逻辑还需要用户自行处理。ReadTimeoutHandler 做了一个简单的处理，就是发生读空闲事件时，就抛出异常。

@Override
    protected final void channelIdle(ChannelHandlerContext ctx, IdleStateEvent evt) throws Exception {
        assert evt.state() == IdleState.READER_IDLE;
        readTimedOut(ctx);
    }

    /**
     * Is called when a read timeout was detected.
     */
    protected void readTimedOut(ChannelHandlerContext ctx) throws Exception {
        if (!closed) {
            // 抛出异常
            ctx.fireExceptionCaught(ReadTimeoutException.INSTANCE);
            ctx.close();
            closed = true;
        }
    }

(5) `WriteTimeoutHandler`

值得注意的是，它什么时候回抛出异常。在类的开头就有一句：

Raises a {@link WriteTimeoutException} when a write operation cannot finish in a certain period of time.

当一个写操作在一段时间没有完成的时候抛出 WriteTimeoutException 。

所以实际上，这是对写操作在一段时间没有完成的时候做的处理。

@Override
        public void run() {
            // 操作是否完成
            // Was not written yet so issue a write timeout
            // The promise itself will be failed with a ClosedChannelException once the close() was issued
            // See https://github.com/netty/netty/issues/2159
            if (!promise.isDone()) {
                try {
                    // 抛出异常
                    writeTimedOut(ctx);
                } catch (Throwable t) {
                    ctx.fireExceptionCaught(t);
                }
            }
            removeWriteTimeoutTask(this);
        }

❤ 转载请注明本文地址或来源，谢谢合作 ❤

center

码恋

ALL YOUR SMILES, ALL MY LIFE.