Netty源码（三）NioEventLoop三部曲

前言

本文将会具体分析NioEventLoop中的thread，它的启动时机，以及所履行的职责。还会分析一些netty的实现细节，比如解决NIO的bug和一些优化等。

thread启动

之前说到NioEventLoop是由一个thread处理I/O事件和提交的任务。先看一下这个thread启动的流程。

execute 简化流程

private void execute(Runnable task, boolean immediate) {
       //是当前线程调用，直接加入队列
        boolean inEventLoop = inEventLoop();
        addTask(task);
        if (!inEventLoop) {
            //启动线程
            startThread();
         // ......
        }

        if (!addTaskWakesUp && immediate) {
            wakeup(inEventLoop);
        }
    }

可以看出启动thread是一个延迟加载的过程，在执行第一个任务的时候才会启动thread。跟进去看startThread()

    private void startThread() {
      //判断线程状态是否已启动
        if (state == ST_NOT_STARTED) {
           //CAS设置线程状态为已启动
            if (STATE_UPDATER.compareAndSet(this, ST_NOT_STARTED, ST_STARTED)) {
                boolean success = false;
                try {
                  //真正去启动线程
                    doStartThread();
                    success = true;
                } finally {
                    if (!success) {
                        STATE_UPDATER.compareAndSet(this, ST_STARTED, ST_NOT_STARTED);
                    }
                }
            }
        }
    }

doStartThread

private void doStartThread() {
        assert thread == null;
        //调用传入参数的executor的execute方法，
        //executor会新建一个线程去执行任务
        executor.execute(new Runnable() {
            @Override
            public void run() {
                //将执行该任务的线程赋值给thread 
                thread = Thread.currentThread();
                if (interrupted) {
                    thread.interrupt();
                }

                boolean success = false;
                updateLastExecutionTime();
                try {
                    //执行任务
                    SingleThreadEventExecutor.this.run();
                    success = true;
                } catch (Throwable t) {
                    logger.warn("Unexpected exception from an event executor: ", t);
                } finally {
                // ......
            }
    }

前文分析了executor为ThreadPerTaskExecutor，执行execute方法时候为新建一个线程去执行任务，NioEventLoop的thread就是在此时赋值。
thread的启动流程简化为，首先thread启动是一个懒加载的过程，在第一次执行任务才会启动。在启动的过程中，会有一个CAS的状态判断当前线程是否已经被启动，如果thread没有启动，则通过传入的executor对象去创建thread对象，并执行SingleThreadEventExecutor.this.run()这个方法。

下面分析SingleThreadEventExecutor.this.run()这个方法，

    /**
     * Run the tasks in the {@link #taskQueue}
     */
    protected abstract void run();

可以看见是一个抽象方法，然后找到文本分析的NioEventLoop对于run的实现，这里做一个将代码做一个简化，只有主要流程

    protected void run() {
        int selectCnt = 0;
        for (; ; ) {
            //1、检测IO事件
            select();
            try {
              //2、处理准备就绪的IO事件
                processSelectedKeys();
            } finally {
                // 3、执行队列里的任务
                final long ioTime = System.nanoTime() - ioStartTime;
                ranTasks = runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
            }

        }
    }

NioEventLoop的职责只有三个，1、检测IO事件；2、处理准备就绪的IO事件；3、执行队列里的任务，用一个死循环去不断执行这三件事情。如之前画的图所示：

run

接下来就着重分析这三个步骤。

select

select步骤的核心是调用通过NIO中的selector的select()方法，返回selector上所监听到IO事件。

                    case SelectStrategy.SELECT:
                        // 获取当前任务队列的延迟时间
                        long curDeadlineNanos = nextScheduledTaskDeadlineNanos();
                        if (curDeadlineNanos == -1L) {
                            curDeadlineNanos = NONE; // nothing on the calendar
                        }
                        nextWakeupNanos.set(curDeadlineNanos);
                        try {
                            //当前任务队列为空，监听IO事件
                            if (!hasTasks()) {
                                strategy = select(curDeadlineNanos);
                            }
                        } finally {
                            // This update is just to help block unnecessary selector wakeups
                            // so use of lazySet is ok (no race condition)
                            nextWakeupNanos.lazySet(AWAKE);
                        }
                        // fall through
                    default:
                    }

select方法

    private int select(long deadlineNanos) throws IOException {
        if (deadlineNanos == NONE) {
            return selector.select();
        }
        // Timeout will only be 0 if deadline is within 5 microsecs
        long timeoutMillis = deadlineToDelayNanos(deadlineNanos + 995000L) / 1000000L;
        return timeoutMillis <= 0 ? selector.selectNow() : selector.select(timeoutMillis);
    }

流程整体比较简单，如果时间参数deadlineNanos为NONE，就调用selector.select()方法，这个方法会一直阻塞直到有IO事件返回。否则再判断deadlineNanos是否小于等于0，如果是调用selectNow()会立即返回当前selector上准备就绪的IO事件，否则调用selector.select(timeoutMillis)方法，会在指定时间内返回，不管是否有IO事件发生。然后跟select()方法，找到实现类io.netty.channel.nio.SelectedSelectionKeySetSelector，

    public int select() throws IOException {
        selectionKeys.reset();
        return delegate.select();
    }

一共有两步操作，第一步是将之前的selectionKeys清空，检测到就绪的IO事件都会放入selectionKeys中，这里表示新的一轮IO循环开始，所以要将之前的清空（selectionKeys后续会在详细介绍）。第二步是调用NIO中的Selector对象的select()，将最后底层的IO实现委托给它。

processSelectedKeys

processSelectedKeys这一步将会处理监测到的IO事件，比如连接、读写的IO操作。

    private void processSelectedKeys() {
        if (selectedKeys != null) {
            processSelectedKeysOptimized();
        } else {
            processSelectedKeysPlain(selector.selectedKeys());
        }
    }

这里有个细节，处理优化过后的selectedKeys还是处理原生的selectedKeys。所谓优化的selectedKeys就是将原生的selectedKeys的HashSet替换成数组实现，提高空间利用率和遍历的效率，待会儿会详细将到是怎么替换的selectedKeys。

然后跟进去看processSelectedKeysOptimized()的具体实现：

    private void processSelectedKeysOptimized() {
        for (int i = 0; i < selectedKeys.size; ++i) {
            final SelectionKey k = selectedKeys.keys[i];
            // null out entry in the array to allow to have it GC'ed once the Channel close
            // See https://github.com/netty/netty/issues/2363
            selectedKeys.keys[i] = null;

            final Object a = k.attachment();

            if (a instanceof AbstractNioChannel) {
                processSelectedKey(k, (AbstractNioChannel) a);
            } else {
                
                NioTask<SelectableChannel> task = (NioTask<SelectableChannel>) a;
                processSelectedKey(k, task);
            }

            if (needsToSelectAgain) {
                // null out entries in the array to allow to have it GC'ed once the Channel close
                // See https://github.com/netty/netty/issues/2363
                selectedKeys.reset(i + 1);

                selectAgain();
                i = -1;
            }
        }
    }

整体流程就是在遍历selectedKeys，将绑定在SelectionKey上的Channel取下来，然后做对应的IO操作，最后再判断是否需要重置selectedKeys。下面我会逐步分析里面的细节

第一： selectedKeys.keys[i] = null;

将SelectionKey取出之后把数组这个位置的地方置为null。为什么这么做？https://github.com/netty/netty/issues/2363描述的很清楚，简单来说就是我们并不会去清空selectedKeys数组，这就会导致在Channel关闭之后，依然会保持SelectionKey的强引用。

selectedKeys.jpg

如上图所示，假如数组原有长度为2，一次高峰期的IO事件导致数组扩容到8，之后新的IO事件的数量又达不到之前数组的位置，为导致上图坐标[6]、[7]位置会长时间持有已经关闭的Channel的引用，所以这里将其置为null，有助于GC。

第二： processSelectedKey

            
            final Object a = k.attachment();
            if (a instanceof AbstractNioChannel) {
                processSelectedKey(k, (AbstractNioChannel) a);
            }

首先是将SelectionKey绑定的属性取下来，判断是否是AbstractNioChannel的类型。这里可以追踪一下netty是什么时候将AbstractNioChannel设置进去的。在AbstractNioChannel的doRegister方法

  //最后一个参数就是att
  selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);

其Channel注册到底层jdk的组件中，然后将AbstractNioChannel作为参数传递进去，后续轮询出IO事件之后，再将AbstractNioChannel取出做后续操作。
具体处理IO事件
processSelectedKey(SelectionKey k, AbstractNioChannel ch)
这里贴一点核心流程，主要是判断当前Channel的操作类型，是连接还是读、写

           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) {
                // Call forceFlush which will also take care of clear the OP_WRITE once there is nothing left to write
                ch.unsafe().forceFlush();
            }
            //读事件
            if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
                unsafe.read();
            }

这里面的内部流程就不具体分析了，大致分为两个部分bossGroup监听的连接事件，将接受到的Channel转交给workGroup，然后workGroup处理读写事件，然后将事件通过ChannelPipeline将事件传播出去。具体细节可以看AbstractNioMessageChannel和AbstractNioByteChannel的read()方法，后续可能会具体分析这里的代码。

第三: needsToSelectAgain

最后一个步骤，重新设置selectedKeys

            if (needsToSelectAgain) {
                // null out entries in the array to allow to have it GC'ed once the Channel close
                // See https://github.com/netty/netty/issues/2363
                selectedKeys.reset(i + 1);
                selectAgain();
                i = -1;
            }

什么时候需要重新select?找到needsToSelectAgain被设置为true的地方，只有唯一的一处cancel

    void cancel(SelectionKey key) {
        key.cancel();
        cancelledKeys ++;
        if (cancelledKeys >= CLEANUP_INTERVAL) {
            cancelledKeys = 0;
            needsToSelectAgain = true;
        }
    }

然后看cancel被调用的地方doDeregister

    protected void doDeregister() throws Exception {
        eventLoop().cancel(selectionKey());
    }

由上面的两部分代码分析可以知道，channel的关闭是通过移除在selector上的注册实现的，同时会把cancelledKeys加一。当达到了阈值CLEANUP_INTERVAL（默认256）后将cancelledKeys重置为0、needsToSelectAgain 为true。
当needsToSelectAgain 为true之后，有两个步骤：
1.selectedKeys清空 -> selectedKeys.reset(i + 1);

    void reset(int start) {
        Arrays.fill(keys, start, size, null);
        size = 0;
    }

再次填充selectedKeys ->selectAgain

    private void selectAgain() {
        needsToSelectAgain = false;
        try {
            selector.selectNow();
        } catch (Throwable t) {
            logger.warn("Failed to update SelectionKeys.", t);
        }
    }

至于为什么需要重新去填充selectedKeys，可能是需要保持selectedKeys里面的Channel都随时保持的是活跃的。

processSelectedKeys到这就分析完了，总共分为三步

遍历selectedKeys
处理IO事件
是否需要重置selectedKeys

ranTasks

现在分析thread的最后一步工作ranTasks，执行队列里的任务。
1. 任务类型
NioEventLoop里的任务类型分为两部分，一个是由taskQueue（MpscUnboundedArrayQueue）存放普通的任务，还有一个scheduledTaskQueue存放定时任务的队列。之前分析过EventLoop继承自ScheduledExecutorService，所以也需要提供执行定时任务的功能，而这里的定时任务是通过PriorityQueue来实现的。（定时任务的实现方式有很多，优先队列只是其中一种）ranTasks执行的任务其实就是两部分的内容，一个是普通队列中的任务和定时队列中的任务。
2. ioRatio
在分析执行细节之前，在提一个很重要的参数ioRatio，代表设置事件循环中I/O所需时间的百分比，意思就是在一次循环中，处理IO事件的时间与处理队列任务所占时间做一个百分比的分配，范围是1到100，当设置为100时，这个参数就失效了，默认参数为50。下面代码就是对ioRatio的使用

                //等于100的时候，参数失效，不再平衡IO事件所占时间的比例
                if (ioRatio == 100) {
                    try {
                        if (strategy > 0) {
                            processSelectedKeys();
                        }
                    } finally {
                        // Ensure we always run tasks.
                        ranTasks = runAllTasks();
                    }
                } else if (strategy > 0) {
                    //开始执行IO事件的时间
                    final long ioStartTime = System.nanoTime();
                    try {
                        processSelectedKeys();
                    } finally {
                        // 获得IO执行总共耗时
                        final long ioTime = System.nanoTime() - ioStartTime;
                        //按照ioRatio计算出将花费多少时间执行ranTasks 
                        ranTasks = runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
                    }
                }

3. runAllTasks

    protected boolean runAllTasks(long timeoutNanos) {
        //将scheduledTaskQueue队列中的任务转移到taskQueue中
        fetchFromScheduledTaskQueue();
        Runnable task = pollTask();
        //任务为空结束
        if (task == null) {
            afterRunningAllTasks();
            return false;
        }
        //计算本次执行任务最迟的时间
        final long deadline = timeoutNanos > 0 ? ScheduledFutureTask.nanoTime() + timeoutNanos : 0;
        long runTasks = 0;
        long lastExecutionTime;
        for (;;) {
            //执行任务
            safeExecute(task);

            runTasks ++;

            //每执行64个任务之后判断时间是否超出，若超出结束循环
            if ((runTasks & 0x3F) == 0) {
                lastExecutionTime = ScheduledFutureTask.nanoTime();
                if (lastExecutionTime >= deadline) {
                    break;
                }
            }
            //没有任务结束循环
            task = pollTask();
            if (task == null) {
                lastExecutionTime = ScheduledFutureTask.nanoTime();
                break;
            }
        }

        afterRunningAllTasks();
        this.lastExecutionTime = lastExecutionTime;
        return true;
    }

主要流程为

scheduledTaskQueue队列中的任务转移到taskQueue中；
安全的执行任务（其实就是将任务try catch，以免任务执行发生异常，影响其他任务执行）；

    protected static void safeExecute(Runnable task) {
        try {
            task.run();
        } catch (Throwable t) {
            logger.warn("A task raised an exception. Task: {}", task, t);
        }
    }

每执行64个任务之后判断执行时间是否超出deadline，这里采用64个任务为一个批次，没有每次任务执行去判断，也是对性能的一个优化；
执行afterRunningAllTasks方法，其实就是执行tailTasks队列中的任务，然后记录一下最后的执行时间this.lastExecutionTime = lastExecutionTime;

一些细节

selectedKeySet

前面提到过netty将NIO中Selector的selectedKeys替换，这里分析一下为什么需要替换和么去替换的selectedKeys。

为什么替换

NIO原生的selectedKeys使用的是HashSet，而NioEventLoop将其替换成了SelectedSelectionKeySet

//SelectorImpl
protected Set<SelectionKey> selectedKeys = new HashSet();
//NioEventLoop
private SelectedSelectionKeySet selectedKeys;

SelectedSelectionKeySet构造函数

    SelectedSelectionKeySet() {
        keys = new SelectionKey[1024];
    }

SelectedSelectionKeySet使用的是数组存储元素，而HashSet是基于HashMap去存储数据，采用数组使得空间利用率和遍历的效率有所提高。

2.怎么替换

要在运行时替换掉类的属性，很明显是通过反射来做到的。

获取sun.nio.ch.SelectorImpl Class对象

        Object maybeSelectorImplClass = AccessController.doPrivileged(new PrivilegedAction<Object>() {
            @Override
            public Object run() {
                try {
                    return Class.forName(
                            "sun.nio.ch.SelectorImpl",
                            false,
                            PlatformDependent.getSystemClassLoader());
                } catch (Throwable cause) {
                    return cause;
                }
            }
        });

创建selectedKeySet

final SelectedSelectionKeySet selectedKeySet = new SelectedSelectionKeySet();

设置属性

    //获取属性
    Field selectedKeysField = selectorImplClass.getDeclaredField("selectedKeys");
    Field publicSelectedKeysField = selectorImplClass.getDeclaredField("publicSelectedKeys");
    ......
    //将selectedKeySet设置到属性中
    selectedKeysField.set(unwrappedSelector, selectedKeySet);
    publicSelectedKeysField.set(unwrappedSelector, selectedKeySet);

NIO空轮询bug

NIO有一个很出名的bug就是epoll空轮询的bug，这会导致CPU占有率到100%，java也并没有修复这个bug，netty采用了一个很巧妙的方法来绕过这个bug。
主要思想就是，通过检测发生空轮询的次数，当超过一定的阈值之后，netty将会重新创建一个selector，并将之前selector上的channel转移到新的selector上。通过重新创建selector的方式来解决NIO空轮询的bug。

unexpectedSelectorWakeup

        //空轮询的次数超过阈值，默认为512
        if (SELECTOR_AUTO_REBUILD_THRESHOLD > 0 &&
                selectCnt >= SELECTOR_AUTO_REBUILD_THRESHOLD) {
            // The selector returned prematurely many times in a row.
            // Rebuild the selector to work around the problem.
            //重新构建selector
            rebuildSelector();
            return true;
        }

跟进去找到具体的实现方法rebuildSelector0

        final Selector oldSelector = selector;
        final SelectorTuple newSelectorTuple;
        try {
            //创建新的selector
            newSelectorTuple = openSelector();
        } catch (Exception e) {
            logger.warn("Failed to create a new Selector.", e);
            return;
        }
        // Register all channels to the new Selector.
        int nChannels = 0;
        for (SelectionKey key: oldSelector.keys()) {
        //将旧的selector上的channel全部注册到新的selector上
        }
        //赋值
        selector = newSelectorTuple.selector;
        unwrappedSelector = newSelectorTuple.unwrappedSelector;
        try {
            // 关闭旧的selector
            oldSelector.close();
        } catch (Throwable t) {
            if (logger.isWarnEnabled()) {
                logger.warn("Failed to close the old Selector.", t);
            }
        }

总结

本文分析NioEventLoop中所对应的唯一的thread，启动是一个懒加载的过程，当第一次任务执行的时候才会初始化。后续thread开始循环处理三件事件

检测IO事件；
处理准备就绪的IO事件；
执行队列里的任务

本文也对具体的代码进行了分析，还有一些netty对NIO的优化和bug处理，当然netty的精妙之处远不止本文分析的这些，更多的还需要自己去探索和学习。