Netty源码------NioEventLoop

目录

Netty源码------NioEventLoop

1、初识 NioEventLoop 

2、NioEventLoop创建

3、NioEventLoop启动

4、NioEventLoop执行过程

5、 Netty 解决JDK 空轮询Bug

6、总结

1、初识 NioEventLoop 

1.1 先来简单回顾一下Netty模型

• Netty中的Channel系列类型，对应于经典Reactor模型中的client， 封装了用户的通讯连接。
• Netty中的EventLoop系列类型，对应于经典Reactor模型中的Reactor，完成Channel的注册、轮询、分发。
• Netty中的Handler系列类型，对应于经典Reactor模型中的Handler，不过Netty中的Handler设计得更加的高级和巧妙，使用了Pipeline模式。

为了简单了解，来个对比吧：

EventLoopGroup bossGroup = new NioEventLoopGroup(1);
ServerBootstrap server = new ServerBootstrap();
server.group(bossGroup);

EventLoopGroup bossGroup = new NioEventLoopGroup(128);
ServerBootstrap server = new ServerBootstrap();
server.group(bossGroup);

EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup);

1.2 再来了解一下NioEventLoop 和本地Selector的对应关系

NioEventLoop类型绑定了两个重要的java本地类型：一个线程类型，一个Selector类型。本地Selector属性的作用，用于注册Java本地channel。本地线程属性的作用，主要是用于轮询。

1.3 NioEventLoopGroup与NioEventLoop的关系

下面这张图其实是对Netty中NioEventLoop执行的流程图：

• 从里面可以知道，NioEventLoop就是在NioEventLoopGroup中创建的，一个NioEventLoopGroup可以包含多个NioEventLoop。
• 它会在Channel注册完之后，调用run()开始工作。详细的内容，下面介绍：

下面进入源码分析： 

2、NioEventLoop创建

从我们常写的代码入手：

public void start() {init();//Netty封装了NIO，Reactor模型，Boss，worker// Boss线程EventLoopGroup bosGroup = new NioEventLoopGroup();//worker工作线程EventLoopGroup workerGroup = new NioEventLoopGroup();try {//相当于Nio中的ServerSocketChannelServerBootstrap server = new ServerBootstrap();//开始设置各种参数server.group(bosGroup, workerGroup)//主线程处理类.channel(NioServerSocketChannel.class)//子线程处理类Handler.childHandler(new ChannelInitializer<SocketChannel>() {protected void initChannel(SocketChannel client) throws Exception {//这里采用了责任链模式进行client.pipeline().addLast(new HttpResponseEncoder()) //编码器.addLast(new HttpRequestDecoder())    //解码器.addLast(new GPTomcatHandler());     //业务逻辑处理}}).option(ChannelOption.SO_BACKLOG, 64)  //针对主线程的配置，分配线程最大数量.childOption(ChannelOption.SO_KEEPALIVE, true); //针对子线程的配置，保吃长连接//启动服务，绑定端口，异步ChannelFuture future = server.bind(port).sync();System.out.println("GP Tomcat 已启动，监听的端口是：" + port);future.channel().closeFuture().sync();} catch (Exception e) {e.printStackTrace();} finally {//关闭线程池bosGroup.shutdownGracefully();workerGroup.shutdownGracefully();}}

从new NioEventLoopGroup() 开始跟踪，我们会依次得到下面的代码：

public NioEventLoopGroup() {this(0);
}public NioEventLoopGroup(int nThreads) {this(nThreads, (Executor) null);
}//线程组默认线程数为2倍的cpu数
//DEFAULT_EVENT_LOOP_THREADS=2*Runtime.getRuntime().availableProcessors()
protected MultithreadEventLoopGroup(int nThreads, Executor executor, Object... args) {super(nThreads == 0 ? DEFAULT_EVENT_LOOP_THREADS : nThreads, executor, args);
}#MultithreadEventExecutorGroup类，核心方法
protected MultithreadEventExecutorGroup(int nThreads, Executor executor,EventExecutorChooserFactory chooserFactory, Object... args) {if (nThreads <= 0) {throw new IllegalArgumentException(String.format("nThreads: %d (expected: > 0)", nThreads));}if (executor == null) {//线程创建器，负责创建NioEventLoopGroup对应底层线程executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());}children = new EventExecutor[nThreads];//创建NioEventLoop对象数组for (int i = 0; i < nThreads; i ++) {boolean success = false;try {//for循环创建每个NioEventLoop,调用newChild()，配置NioEventLoop核心参数children[i] = newChild(executor, args);//newChildsuccess = true;} catch (Exception e) {// TODO: Think about if this is a good exception typethrow new IllegalStateException("failed to create a child event loop", e);} finally {if (!success) {for (int j = 0; j < i; j ++) {children[j].shutdownGracefully();}for (int j = 0; j < i; j ++) {EventExecutor e = children[j];try {while (!e.isTerminated()) {e.awaitTermination(Integer.MAX_VALUE, TimeUnit.SECONDS);}} catch (InterruptedException interrupted) {// Let the caller handle the interruption.Thread.currentThread().interrupt();break;}}}}}//线程选择器,给每个新连接分配NioEventLoop线程chooser = chooserFactory.newChooser(children);final FutureListener<Object> terminationListener = new FutureListener<Object>() {@Overridepublic void operationComplete(Future<Object> future) throws Exception {if (terminatedChildren.incrementAndGet() == children.length) {terminationFuture.setSuccess(null);}}};for (EventExecutor e: children) {e.terminationFuture().addListener(terminationListener);}Set<EventExecutor> childrenSet = new LinkedHashSet<EventExecutor>(children.length);Collections.addAll(childrenSet, children);readonlyChildren = Collections.unmodifiableSet(childrenSet);}#使用threadFactory创建线程
public final class ThreadPerTaskExecutor implements Executor {private final ThreadFactory threadFactory;public ThreadPerTaskExecutor(ThreadFactory threadFactory) {if (threadFactory == null) {throw new NullPointerException("threadFactory");}this.threadFactory = threadFactory;}@Overridepublic void execute(Runnable command) {//每次执行任务创建一个线程，newDefaultThreadFactory定义了nioEventLoop-1-xx的线程名threadFactory.newThread(command).start();}
}

从上面我们可以看出创建NioEventLoopGroup的大致过程如下：

• 如果没有传入nThread,nThread 默认为CPU核心数的两倍
• 线程创建器，负责创建NioEventLoopGroup对应底层线程
• 创建NioEventLoop对象数组，并配置好其的核心参数

接下来，我们进入newChild()方法，也就是配置NioEventLoop核心参数，看看他到底干了些啥事。

#NioEventLoopGroup
@Overrideprotected EventLoop newChild(Executor executor, Object... args) throws Exception {return new NioEventLoop(this, executor, (SelectorProvider) args[0],((SelectStrategyFactory) args[1]).newSelectStrategy(), (RejectedExecutionHandler) args[2]);}NioEventLoop(NioEventLoopGroup parent, Executor executor, SelectorProvider selectorProvider,SelectStrategy strategy, RejectedExecutionHandler rejectedExecutionHandler) {super(parent, executor, false, DEFAULT_MAX_PENDING_TASKS, rejectedExecutionHandler);//父类构造函数if (selectorProvider == null) {throw new NullPointerException("selectorProvider");}if (strategy == null) {throw new NullPointerException("selectStrategy");}provider = selectorProvider;final SelectorTuple selectorTuple = openSelector();selector = selectorTuple.selector;unwrappedSelector = selectorTuple.unwrappedSelector;selectStrategy = strategy;}#SingleThreadEventExecutor类，父类构造函数
protected SingleThreadEventExecutor(EventExecutorGroup parent, Executor executor,boolean addTaskWakesUp, int maxPendingTasks,RejectedExecutionHandler rejectedHandler) {super(parent);this.addTaskWakesUp = addTaskWakesUp;this.maxPendingTasks = Math.max(16, maxPendingTasks);this.executor = ObjectUtil.checkNotNull(executor, "executor");taskQueue = newTaskQueue(this.maxPendingTasks);//task队列，外部线程将任务扔进队列rejectedExecutionHandler = ObjectUtil.checkNotNull(rejectedHandler, "rejectedHandler");}//task queue
protected Queue<Runnable> newTaskQueue(int maxPendingTasks) {// This event loop never calls takeTask()return PlatformDependent.newMpscQueue(maxPendingTasks);
}private SelectorTuple openSelector() {final Selector unwrappedSelector;try {unwrappedSelector = provider.openSelector();} catch (IOException e) {throw new ChannelException("failed to open a new selector", e);}if (DISABLE_KEYSET_OPTIMIZATION) {return new SelectorTuple(unwrappedSelector);}final SelectedSelectionKeySet selectedKeySet = new SelectedSelectionKeySet();//用数组实现setObject maybeSelectorImplClass = AccessController.doPrivileged(new PrivilegedAction<Object>() {@Overridepublic Object run() {try {return Class.forName("sun.nio.ch.SelectorImpl",false,PlatformDependent.getSystemClassLoader());} catch (Throwable cause) {return cause;}}});if (!(maybeSelectorImplClass instanceof Class) ||// ensure the current selector implementation is what we can instrument.!((Class<?>) maybeSelectorImplClass).isAssignableFrom(unwrappedSelector.getClass())) {if (maybeSelectorImplClass instanceof Throwable) {Throwable t = (Throwable) maybeSelectorImplClass;logger.trace("failed to instrument a special java.util.Set into: {}", unwrappedSelector, t);}return new SelectorTuple(unwrappedSelector);}final Class<?> selectorImplClass = (Class<?>) maybeSelectorImplClass;//通过反射方式设置selectedKeySetObject maybeException = AccessController.doPrivileged(new PrivilegedAction<Object>() {@Overridepublic Object run() {try {Field selectedKeysField = selectorImplClass.getDeclaredField("selectedKeys");Field publicSelectedKeysField = selectorImplClass.getDeclaredField("publicSelectedKeys");Throwable cause = ReflectionUtil.trySetAccessible(selectedKeysField);if (cause != null) {return cause;}cause = ReflectionUtil.trySetAccessible(publicSelectedKeysField);if (cause != null) {return cause;}selectedKeysField.set(unwrappedSelector, selectedKeySet);publicSelectedKeysField.set(unwrappedSelector, selectedKeySet);return null;} catch (NoSuchFieldException e) {return e;} catch (IllegalAccessException e) {return e;}}});if (maybeException instanceof Exception) {selectedKeys = null;Exception e = (Exception) maybeException;logger.trace("failed to instrument a special java.util.Set into: {}", unwrappedSelector, e);return new SelectorTuple(unwrappedSelector);}selectedKeys = selectedKeySet;logger.trace("instrumented a special java.util.Set into: {}", unwrappedSelector);return new SelectorTuple(unwrappedSelector,new SelectedSelectionKeySetSelector(unwrappedSelector, selectedKeySet));}

     进行NioEventLoop参数初始化时，干了很多事，但我们重点知道：

• 创建一个selector，并调用provider.openSelector()创建selector，轮询初始化连接；
• 创建了一个线程阻塞队列BlockIngqueue<Runable>

接下来我们在回到newChild()方法之后会调用的chooserFactory.newChooser(children);看看它又干了什么大事。

#DefaultEventExecutorChooserFactory 类
public final class DefaultEventExecutorChooserFactory implements EventExecutorChooserFactory {public static final DefaultEventExecutorChooserFactory INSTANCE = new DefaultEventExecutorChooserFactory();private DefaultEventExecutorChooserFactory() { }@SuppressWarnings("unchecked")@Overridepublic EventExecutorChooser newChooser(EventExecutor[] executors) {if (isPowerOfTwo(executors.length)) {return new PowerOfTwoEventExecutorChooser(executors);} else {return new GenericEventExecutorChooser(executors);}}//判断长度是否是2的幂，是则使用PowerOfTwoEventExecutorChooser,更高效private static boolean isPowerOfTwo(int val) {return (val & -val) == val;}private static final class PowerOfTwoEventExecutorChooser implements EventExecutorChooser {private final AtomicInteger idx = new AtomicInteger();private final EventExecutor[] executors;PowerOfTwoEventExecutorChooser(EventExecutor[] executors) {this.executors = executors;}@Overridepublic EventExecutor next() {return executors[idx.getAndIncrement() & executors.length - 1];//&比取模高效，循环下标}}private static final class GenericEventExecutorChooser implements EventExecutorChooser {private final AtomicInteger idx = new AtomicInteger();private final EventExecutor[] executors;GenericEventExecutorChooser(EventExecutor[] executors) {this.executors = executors;}@Overridepublic EventExecutor next() {return executors[Math.abs(idx.getAndIncrement() % executors.length)];//取模}}
}

上面的代码逻辑主要表达的意思是：即如果nThreads 是2 的幂，则使用PowerOfTwoEventExecutorChooser，否则使用GenericEventExecutorChooser。这两个Chooser 都重写next()方法。next()方法的主要功能就是将数组索引循环位移，如下图所示：

　　当索引移动最后一个位置时，再调用next()方法就会将索引位置重新指向0。

　　这个运算逻辑其实很简单，就是每次让索引自增后和数组长度取模：idx.getAndIncrement() % executors.length。但是就连一个非常简单的数组索引运算，Netty 都帮我们做了优化。因为在计算机底层，&与比%运算效率更高。

     最后，来张时序图总结一下：

3、NioEventLoop启动

NioEventLoop的启动此时是在Channel初始化，并在绑定端口之后开始的。具体可以回顾上篇博客：

---

Channel的创建：https://blog.csdn.net/qqq3117004957/article/details/106440866#Channel%E6%B3%A8%E5%86%8C

---

#AbstractBootstrap类doBind方法
private ChannelFuture doBind(final SocketAddress localAddress) {final ChannelFuture regFuture = initAndRegister();final Channel channel = regFuture.channel();if (regFuture.cause() != null) {return regFuture;}if (regFuture.isDone()) {// At this point we know that the registration was complete and successful.ChannelPromise promise = channel.newPromise();doBind0(regFuture, channel, localAddress, promise);return promise;} else {// Registration future is almost always fulfilled already, but just in case it's not.final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel);regFuture.addListener(new ChannelFutureListener() {@Overridepublic void operationComplete(ChannelFuture future) throws Exception {Throwable cause = future.cause();if (cause != null) {// Registration on the EventLoop failed so fail the ChannelPromise directly to not cause an// IllegalStateException once we try to access the EventLoop of the Channel.promise.setFailure(cause);} else {// Registration was successful, so set the correct executor to use.// See https://github.com/netty/netty/issues/2586promise.registered();doBind0(regFuture, channel, localAddress, promise);}}});return promise;}}#AbstractBootstrap类doBind0方法
private static void doBind0(final ChannelFuture regFuture, final Channel channel,final SocketAddress localAddress, final ChannelPromise promise) {// This method is invoked before channelRegistered() is triggered.  Give user handlers a chance to set up// the pipeline in its channelRegistered() implementation.//调用SingleThreadEventExecutor中execute方法channel.eventLoop().execute(new Runnable() {@Overridepublic void run() {if (regFuture.isSuccess()) {channel.bind(localAddress, promise).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);} else {promise.setFailure(regFuture.cause());}}});}#SingleThreadEventExecutor中execute方法
public void execute(Runnable task) {if (task == null) {throw new NullPointerException("task");}boolean inEventLoop = inEventLoop();//判断是否当前eventloop中if (inEventLoop) {addTask(task);} else {startThread();//创建线程addTask(task);if (isShutdown() && removeTask(task)) {reject();}}if (!addTaskWakesUp && wakesUpForTask(task)) {wakeup(inEventLoop);}}public boolean inEventLoop() {return inEventLoop(Thread.currentThread());}public boolean inEventLoop(Thread thread) {return thread == this.thread;}private void startThread() {if (STATE_UPDATER.get(this) == ST_NOT_STARTED) {if (STATE_UPDATER.compareAndSet(this, ST_NOT_STARTED, ST_STARTED)) {doStartThread();}}}//启动线程  
private void doStartThread() {assert thread == null;executor.execute(new Runnable() {@Overridepublic void run() {thread = Thread.currentThread();//保存当前线程，用于判断if (interrupted) {thread.interrupt();}boolean success = false;updateLastExecutionTime();try {SingleThreadEventExecutor.this.run();//触发NioEventLoop执行success = true;} catch (Throwable t) {logger.warn("Unexpected exception from an event executor: ", t);} finally {for (;;) {int oldState = STATE_UPDATER.get(SingleThreadEventExecutor.this);if (oldState >= ST_SHUTTING_DOWN || STATE_UPDATER.compareAndSet(SingleThreadEventExecutor.this, oldState, ST_SHUTTING_DOWN)) {break;}}// Check if confirmShutdown() was called at the end of the loop.if (success && gracefulShutdownStartTime == 0) {logger.error("Buggy " + EventExecutor.class.getSimpleName() + " implementation; " +SingleThreadEventExecutor.class.getSimpleName() + ".confirmShutdown() must be called " +"before run() implementation terminates.");}try {// Run all remaining tasks and shutdown hooks.for (;;) {if (confirmShutdown()) {break;}}} finally {try {cleanup();} finally {STATE_UPDATER.set(SingleThreadEventExecutor.this, ST_TERMINATED);threadLock.release();if (!taskQueue.isEmpty()) {logger.warn("An event executor terminated with " +"non-empty task queue (" + taskQueue.size() + ')');}terminationFuture.setSuccess(null);}}}}});}

NioEventLoop启动流程步骤：

• bind->execute(task)[入口]:调用NioEventLoop的execute()方法执行绑定端口，操作封装的Task
• startThread()->doStartThread()[创建线程]:非NioEventLoop线程调用startThread()方法，创建启动线程
• ThreadPerTaskExecutor.execute():线程执行器执行任务，创建并启动FastThreadLocalThread线程
• NioEventLoop.run()[启动]

4、NioEventLoop执行过程

在上面doStartThread()方法中最终调用的是SingleThreadEventExecutor.this.run()方法，这个this 就是NioEventLoop 对象：

protected void run() {for (;;) {try {try {switch (selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())) {case SelectStrategy.CONTINUE:continue;case SelectStrategy.BUSY_WAIT:case SelectStrategy.SELECT:select(wakenUp.getAndSet(false));if (wakenUp.get()) {selector.wakeup();}default:}cancelledKeys = 0;needsToSelectAgain = false;final int ioRatio = this.ioRatio;if (ioRatio == 100) {try {processSelectedKeys();} finally {runAllTasks();}　　　　　　　　　　......}}

终于看到似曾相识的代码。上面代码主要就是用一个死循环，在不断地轮询SelectionKey.select()方法，主要用来解决JDK 空轮询Bug，而processSelectedKeys()就是针对不同的轮询事件进行处理。如果客户端有数据写入，最终也会调用AbstractNioMessageChannel 的doReadMessages()方法。那么我们先来看一下是哪里调用的，通过追踪processSelectedKeys()方法，最后会调用到NioEventLoop的processSelectedKey 方法：

private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();if (!k.isValid()) {final EventLoop eventLoop;eventLoop = ch.eventLoop();
int readyOps = k.readyOps();if ((readyOps & SelectionKey.OP_CONNECT) != 0) {int ops = k.interestOps();ops &= ~SelectionKey.OP_CONNECT;k.interestOps(ops);unsafe.finishConnect();}if ((readyOps & SelectionKey.OP_WRITE) != 0) {ch.unsafe().forceFlush();}if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {unsafe.read();}} catch (CancelledKeyException ignored) {unsafe.close(unsafe.voidPromise());}}

　　这里可以看到熟悉的代码，当有链接进来的时候，便会走 unsafe.read()：这个里面就调用了doReadMessages

1. Connect, 即连接事件(TCP 连接), 对应于SelectionKey.OP_CONNECT.int值为16.
2. Accept, 即确认事件, 对应于SelectionKey.OP_ACCEPT.int值为8.
3. Read, 即读事件, 对应于SelectionKey.OP_READ, 表示 buffer 可读.int值为1
4. Write, 即写事件, 对应于SelectionKey.OP_WRITE, 表示 buffer 可写.int值为4

总结一下：

1. Netty 中Selector 事件轮询是从EventLoop 的execute()方法开始的。
2. 在EventLoop 的execute()方法中，会为每一个任务创建一个独立的线程，并保存到无锁化串行任务队列。
3. 线程任务队列的每个任务实际调用的是NioEventLoop 的run()方法。
4. 在run 方法中调用processSelectedKeys()处理轮询事件。

5、 Netty 解决JDK 空轮询Bug

各位应该早有耳闻臭名昭著的Java NIO epoll 的bug，它会导致Selector 空轮询，最终导致CPU 100%。官方声称在JDK1.6 版本的update18 修复了该问题，但是直到JDK1.7 版本该问题仍旧存在，只不过该BUG 发生概率降低了一些而已，它并没有被根本解决。出现此Bug 是因为当Selector 的轮询结果为空，也没有wakeup 或新消息处理，则发生空轮询，CPU 使用率达到100%。

在Netty 中最终的解决办法是：创建一个新的Selector，将可用事件重新注册到新的Selector 中来终止空轮训。前面我们有提到select()方法解决了JDK 空轮训的Bug，它到底是如何解决的呢？下面我们来一探究竟，进入select()方法的源码：

private void select(boolean oldWakenUp) throws IOException {Selector selector = this.selector;try {int selectCnt = 0;long currentTimeNanos = System.nanoTime();long selectDeadLineNanos = currentTimeNanos + delayNanos(currentTimeNanos);for (;;) {　　　　　　　　　 // .......int selectedKeys = selector.select(timeoutMillis);selectCnt ++;if (selectedKeys != 0 || oldWakenUp || wakenUp.get() || hasTasks() || hasScheduledTasks()) {break;}if (Thread.interrupted()) {//.....selectCnt = 1;break;}long time = System.nanoTime();if (time - TimeUnit.MILLISECONDS.toNanos(timeoutMillis) >= currentTimeNanos) {// timeoutMillis elapsed without anything selected.selectCnt = 1;} else if (SELECTOR_AUTO_REBUILD_THRESHOLD > 0 &&selectCnt >= SELECTOR_AUTO_REBUILD_THRESHOLD) {selector = selectRebuildSelector(selectCnt);selectCnt = 1;break;}currentTimeNanos = time;}}

从上面的代码中可以看出,Selector 每一次轮询都计数selectCnt++，开始轮询会计时赋值给timeoutMillis，轮询完成会计时赋值给time，这两个时间差会有一个时间差，而这个时间差就是每次轮询所消耗的时间。从上面的的逻辑看出，如果每次轮询消耗的时间为0，且重复次数超过512 次，则调用rebuildSelector()方法，即重构Selector。

继续跟进源码： 

private void rebuildSelector0() {final Selector oldSelector = selector;final SelectorTuple newSelectorTuple;newSelectorTuple = openSelector();// Register all channels to the new Selector.int nChannels = 0;for (SelectionKey key: oldSelector.keys()) {Object a = key.attachment();try {if (!key.isValid() || key.channel().keyFor(newSelectorTuple.unwrappedSelector) != null) {continue;}int interestOps = key.interestOps();key.cancel();SelectionKey newKey = key.channel().register(newSelectorTuple.unwrappedSelector, interestOps, a);}

在rebuildSelector()方法中，主要做了三件事情：

1. 创建一个新的Selector。
2. 将原来Selector 中注册的事件全部取消。
3. 将可用事件重新注册到新的Selector 中，并激活。

6、总结

说实话，自己总结了这么长，看完都得花半个小时以上，但我认为了解源码并不是为了记住它，而是从中了解Netty的工作机制，有一定的印象就为成功。所以来个小总结：

• NioEventLoop就相当于NIO编程中的Reactor;
• 默认的一个NioEventLopGroup会创建CPU核心数的2倍线，即NioEventLoop的数量；
• NioEventLoop是在Channel注册好之后，调用doBind0（）方法开始的；

 

Netty空轮询的解决方法：

netty 会在每次进行 selector.select(timeoutMillis) 之前记录一下开始时间currentTimeNanos，在select之后记录一下结束时间，判断select操作是否至少持续了timeoutMillis秒（这里将time - TimeUnit.MILLISECONDS.toNanos(timeoutMillis) >= currentTimeNanos改成time - currentTimeNanos >= TimeUnit.MILLISECONDS.toNanos(timeoutMillis)或许更好理解一些）, 如果持续的时间大于等于timeoutMillis，说明就是一次有效的轮询，重置selectCnt标志，否则，表明该阻塞方法并没有阻塞这么长时间，可能触发了jdk的空轮询bug，当空轮询的次数超过一个阀值的时候，默认是512，就开始重建selector。