netty源码阅读之NioEventLoop之NioEventLoop执行----select()检查io事件

不念不忘少年蓝@ 2022-05-17 10:34 279阅读 0赞

从《[netty源码阅读之NioEventLoop之NioEventLoop执行][netty_NioEventLoop_NioEventLoop]》的select(wakenUp.getAndSet(false))这个函数开始分析，select的时候，先把wakenUp这个标志位设置为false，也就是现在没有用户唤醒了，并把之前是否被用户唤醒作为oldWakenUp传入，方便以后使用，进入之后的代码为：

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 (;;) {
                    long timeoutMillis = (selectDeadLineNanos - currentTimeNanos + 500000L) / 1000000L;
                    if (timeoutMillis <= 0) {
                        if (selectCnt == 0) {
                            selector.selectNow();
                            selectCnt = 1;
                        }
                        break;
                    }
    
                    // If a task was submitted when wakenUp value was true, the task didn't get a chance to call
                    // Selector#wakeup. So we need to check task queue again before executing select operation.
                    // If we don't, the task might be pended until select operation was timed out.
                    // It might be pended until idle timeout if IdleStateHandler existed in pipeline.
                    if (hasTasks() && wakenUp.compareAndSet(false, true)) {
                        selector.selectNow();
                        selectCnt = 1;
                        break;
                    }
    
                    int selectedKeys = selector.select(timeoutMillis);
                    selectCnt ++;
    
                    if (selectedKeys != 0 || oldWakenUp || wakenUp.get() || hasTasks() || hasScheduledTasks()) {
                        // - Selected something,
                        // - waken up by user, or
                        // - the task queue has a pending task.
                        // - a scheduled task is ready for processing
                        break;
                    }
                    if (Thread.interrupted()) {
                        // Thread was interrupted so reset selected keys and break so we not run into a busy loop.
                        // As this is most likely a bug in the handler of the user or it's client library we will
                        // also log it.
                        //
                        // See https://github.com/netty/netty/issues/2426
                        if (logger.isDebugEnabled()) {
                            logger.debug("Selector.select() returned prematurely because " +
                                    "Thread.currentThread().interrupt() was called. Use " +
                                    "NioEventLoop.shutdownGracefully() to shutdown the NioEventLoop.");
                        }
                        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) {
                        // The selector returned prematurely many times in a row.
                        // Rebuild the selector to work around the problem.
                        logger.warn(
                                "Selector.select() returned prematurely {} times in a row; rebuilding Selector {}.",
                                selectCnt, selector);
    
                        rebuildSelector();
                        selector = this.selector;
    
                        // Select again to populate selectedKeys.
                        selector.selectNow();
                        selectCnt = 1;
                        break;
                    }
    
                    currentTimeNanos = time;
                }
    
                if (selectCnt > MIN_PREMATURE_SELECTOR_RETURNS) {
                    if (logger.isDebugEnabled()) {
                        logger.debug("Selector.select() returned prematurely {} times in a row for Selector {}.",
                                selectCnt - 1, selector);
                    }
                }
            } catch (CancelledKeyException e) {
                if (logger.isDebugEnabled()) {
                    logger.debug(CancelledKeyException.class.getSimpleName() + " raised by a Selector {} - JDK bug?",
                            selector, e);
                }
                // Harmless exception - log anyway
            }
        }

很长，我们分三步来解析这个源码：

1、检查定时任务是否到期，检查是否有任务执行，二者有一就进行非阻塞select（jdk原生select）

2、如果上面都没有，则进行一定时间的阻塞式select，也就是int selectedKeys = selector.select(timeoutMillis);

3、最后判断是否空轮询，是否达到空轮询的次数，如果达到，就把重建selector，这个selector就有可能不会有空轮询的情况发生了。

下面我们一个个分析：

1、

`long selectDeadLineNanos = currentTimeNanos + delayNanos(currentTimeNanos);返回的是定时任务队列的最近的需要执行的时间（这个定时任务队列，和我们之前的mpsc队列不同，后面会讲到），如果这个时间减去当前时间小于0，那么就说明有定时任务要执行，直接非阻塞select后退出，并且把selectCnt 置为1（后面会讲解）;`

// If a task was submitted when wakenUp value was true, the task didn't get a chance to call
                    // Selector#wakeup. So we need to check task queue again before executing select operation.
                    // If we don't, the task might be pended until select operation was timed out.
                    // It might be pended until idle timeout if IdleStateHandler existed in pipeline.
                    if (hasTasks() && wakenUp.compareAndSet(false, true)) {
                        selector.selectNow();
                        selectCnt = 1;
                        break;
                    }

这个是说明任务队列有任务，和上执行的操作和上面的一样。

2、

如果前面两步都没有，就进行特定时间的阻塞式等待：

int selectedKeys = selector.select(timeoutMillis);
                    selectCnt ++;

并且把selectCnt加1，方便后面使用。

一定时间的阻塞式等待之后，如果发生了以下时间，都退出select：

if (selectedKeys != 0 || oldWakenUp || wakenUp.get() || hasTasks() || hasScheduledTasks()) {
                        // - Selected something,
                        // - waken up by user, or
                        // - the task queue has a pending task.
                        // - a scheduled task is ready for processing
                        break;
                    }

3、

以下这段代码判断是否有空轮询发生

long time = System.nanoTime();
                    if (time - TimeUnit.MILLISECONDS.toNanos(timeoutMillis) >= currentTimeNanos) {
                        // timeoutMillis elapsed without anything selected.
                        selectCnt = 1;
                    }

我们换一种方式解释，把currentTimeNonos放到左边来。也就是当前时间减去进来这个方法的时间（阻塞之前的时间），大于阻塞的时间的话，那么就没有发生空轮询，selectCnt重置为1；否则，本来应该阻塞这么长时间，没有阻塞，就是发生了空轮询。到这里就明白selectCnt可以简单的解释为空轮询的次数吧，因为如果不发生空轮询，selectCnt会重置为1（或者操作了其他任务）。

发生空轮询到达一定的次数SELECTOR\_AUTO\_REBUILD\_THRESHOLD（查看源码可以知道是512），就会重建selector，并把selectCnt次数重置为1

else 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.
                        logger.warn(
                                "Selector.select() returned prematurely {} times in a row; rebuilding Selector {}.",
                                selectCnt, selector);
    
                        rebuildSelector();
                        selector = this.selector;
    
                        // Select again to populate selectedKeys.
                        selector.selectNow();
                        selectCnt = 1;
                        break;
                    }

查看rebuildSelector()

public void rebuildSelector() {
            if (!inEventLoop()) {
                execute(new Runnable() {
                    @Override
                    public void run() {
                        rebuildSelector();
                    }
                });
                return;
            }
    
            final Selector oldSelector = selector;
            final Selector newSelector;
    
            if (oldSelector == null) {
                return;
            }
    
            try {
                newSelector = 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 (;;) {
                try {
                    for (SelectionKey key: oldSelector.keys()) {
                        Object a = key.attachment();
                        try {
                            if (!key.isValid() || key.channel().keyFor(newSelector) != null) {
                                continue;
                            }
    
                            int interestOps = key.interestOps();
                            key.cancel();
                            SelectionKey newKey = key.channel().register(newSelector, interestOps, a);
                            if (a instanceof AbstractNioChannel) {
                                // Update SelectionKey
                                ((AbstractNioChannel) a).selectionKey = newKey;
                            }
                            nChannels ++;
                        } catch (Exception e) {
                            logger.warn("Failed to re-register a Channel to the new Selector.", e);
                            if (a instanceof AbstractNioChannel) {
                                AbstractNioChannel ch = (AbstractNioChannel) a;
                                ch.unsafe().close(ch.unsafe().voidPromise());
                            } else {
                                @SuppressWarnings("unchecked")
                                NioTask<SelectableChannel> task = (NioTask<SelectableChannel>) a;
                                invokeChannelUnregistered(task, key, e);
                            }
                        }
                    }
                } catch (ConcurrentModificationException e) {
                    // Probably due to concurrent modification of the key set.
                    continue;
                }
    
                break;
            }
    
            selector = newSelector;
    
            try {
                // time to close the old selector as everything else is registered to the new one
                oldSelector.close();
            } catch (Throwable t) {
                if (logger.isWarnEnabled()) {
                    logger.warn("Failed to close the old Selector.", t);
                }
            }
    
            logger.info("Migrated " + nChannels + " channel(s) to the new Selector.");
        }

其实就是把所有之前绑定的东西都重新绑定到新的selector上面。

netty通过这个方式，巧妙地解决了jdk空轮询的bug。

[netty_NioEventLoop_NioEventLoop]: https://blog.csdn.net/fst438060684/article/details/81558585