Netty源码解读(三)-NioEventLoop

2022年6月16日 355点热度 1人点赞 0条评论

先看看EventLoop类图

NioEventLoop.png

我们在Netty第二篇文章中的代码中,看到有多次用到eventLoop.execute()方法,这个方法就是EventLoop开启线程执行任务的关键,跟踪进去看看

// 类SingleThreadEventExecutor
SingleThreadEventExecutor#execute(Runnable task)
-->
SingleThreadEventExecutor#execute0(@Schedule Runnable task)
-->
private void execute(Runnable task, boolean immediate) {
    // 判断当前线程是否为eventLoop的线程
    boolean inEventLoop = inEventLoop();
    // 将任务添加进taskQueue
    addTask(task);
    if (!inEventLoop) {
        // 开启eventLoop的线程
        startThread();
        if (isShutdown()) {
            boolean reject = false;
            try {
                if (removeTask(task)) {
                    reject = true;
                }
            } catch (UnsupportedOperationException e) {
                // The task queue does not support removal so the best thing we can do is to just move on and
                // hope we will be able to pick-up the task before its completely terminated.
                // In worst case we will log on termination.
            }
            if (reject) {
                reject();
            }
        }
    }

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

这段代码,我们分3部分解读

添加任务

将任务添加进队列,等待调用

对应代码

// addTask(task);
// 类SingleThreadEventExecutor
protected void addTask(Runnable task) {
    ObjectUtil.checkNotNull(task, "task");
    // 添加任务
    if (!offerTask(task)) {
        // 如果添加失败,执行拒绝执行处理器
        reject(task);
    }
}

SingleThreadEventExecutor有一个类变量来存储task

private final Queue<Runnable> taskQueue;

开启线程

开启线程是以下这段逻辑

// 判断当前线程是否eventLoop的线程
// 如果不是,则开启EventLoop的线程
if (!inEventLoop) {
    // 开启eventLoop的线程
    startThread();
    。。。。。。
}
-->
// 类SingleThreadEventExecutor
private void startThread() {
    // 一些状态判断,保证doStartThread只会被执行一次
    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);
                }
            }
        }
    }
}
-->
// 类SingleThreadEventExecutor
private void doStartThread() {
    assert thread == null;
    // 这个Executor在EventLoopGroup构造时,就已经注入
    // MultithreadEventExecutorGroup
    // executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());
    executor.execute(new Runnable() {
        @Override
        public void run() {
            thread = Thread.currentThread();
            if (interrupted) {
                thread.interrupt();
            }

            boolean success = false;
            updateLastExecutionTime();
            try {
                // 这个整个NioEventLoop的核心,里面是个死循环
                SingleThreadEventExecutor.this.run();
                success = true;
            } catch (Throwable t) {
                logger.warn("Unexpected exception from an event executor: ", t);
            } finally {
                。。。。。。
            }
        }
    });
}

在前面EventLoopGroup的创建与初始化一节有说到,executor的实例化

executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());
executor.execute(new Runnable() {...});
// 不深究细节的话,上面两行代码效果类比下面两行
Thread thread = new Thread(new Runnable() {...});
thread.start();

所以,executor.execute这里,就开了一个新的线程。

而这个线程,主要处理的是SingleThreadEventExecutor.**this**.run();,也就是NioEventLoop#run()

那这个run方法是干嘛的,看下图红框部分

Untitled

因为run方法比较重要,此处不做代码省略,看注释

protected void run() {
    int selectCnt = 0;
    for (;;) {
        try {
            int strategy;
            try {
                // 如果没有任务,则返回SELECT
                // 有任务,则获取io事件的个数。此时如果strategy >= 0
                // 如果有io任务,优先io任务,然后才执行普通任务
                strategy = selectStrategy.calculateStrategy(selectNowSupplier, hasTasks());
                switch (strategy) {
                case SelectStrategy.CONTINUE:
                    continue;

                case SelectStrategy.BUSY_WAIT:
                    // fall-through to SELECT since the busy-wait is not supported with NIO

                case SelectStrategy.SELECT:
                    // 下一次定时任务触发截止时间
                    long curDeadlineNanos = nextScheduledTaskDeadlineNanos();
                    if (curDeadlineNanos == -1L) {
                        curDeadlineNanos = NONE; // nothing on the calendar
                    }
                    nextWakeupNanos.set(curDeadlineNanos);
                    try {
                        // 再次判断没有任务
                        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:
                }
            } catch (IOException e) {
                // If we receive an IOException here its because the Selector is messed up. Let's rebuild
                // the selector and retry. https://github.com/netty/netty/issues/8566
                rebuildSelector0();
                selectCnt = 0;
                handleLoopException(e);
                continue;
            }

            selectCnt++;
            cancelledKeys = 0;
            needsToSelectAgain = false;
            // 控制处理io事件的事件占用比例,默认是百分之50,一半时间用来处理io事件,一半时间用来处理任务
            final int ioRatio = this.ioRatio;
            boolean ranTasks;
            // 100%表示执行完全部任务,才进入下一轮循环
            if (ioRatio == 100) {
                try {
                    if (strategy > 0) {
                        // 在对应的 Channel 上处理 IO 事件
                        processSelectedKeys();
                    }
                } finally {
                    // Ensure we always run tasks.
                    // 执行queueTask中全部的任务
                    ranTasks = runAllTasks();
                }
            } else if (strategy > 0) {
                final long ioStartTime = System.nanoTime();
                try {
                    // 在对应的 Channel 上处理 IO 事件
                    processSelectedKeys();
                } finally {
                    // Ensure we always run tasks.
                    final long ioTime = System.nanoTime() - ioStartTime;
                    ranTasks = runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
                }
            } else {
                // 0表示运行运行最小数量的任务,即63个
                ranTasks = runAllTasks(0); // This will run the minimum number of tasks
            }

            if (ranTasks || strategy > 0) {
                if (selectCnt > MIN_PREMATURE_SELECTOR_RETURNS && logger.isDebugEnabled()) {
                    logger.debug("Selector.select() returned prematurely {} times in a row for Selector {}.",
                            selectCnt - 1, selector);
                }
                selectCnt = 0;
            } else if (unexpectedSelectorWakeup(selectCnt)) { // Unexpected wakeup (unusual case)
                selectCnt = 0;
            }
        } catch (CancelledKeyException e) {
            // Harmless exception - log anyway
            if (logger.isDebugEnabled()) {
                logger.debug(CancelledKeyException.class.getSimpleName() + " raised by a Selector {} - JDK bug?",
                        selector, e);
            }
        } catch (Error e) {
            throw e;
        } catch (Throwable t) {
            handleLoopException(t);
        } finally {
            // Always handle shutdown even if the loop processing threw an exception.
            try {
                if (isShuttingDown()) {
                    closeAll();
                    if (confirmShutdown()) {
                        return;
                    }
                }
            } catch (Error e) {
                throw e;
            } catch (Throwable t) {
                handleLoopException(t);
            }
        }
    }
}

run里面是个死循环,关键方法有selectprocessSelectedKeysrunAllTasks

  • select

    select方法很简单,就是选择合适的阻塞时间,等待IO事件触发

    private int select(long deadlineNanos) throws IOException {
        if (deadlineNanos == NONE) {
            return selector.select();
        }
        // Timeout will only be 0 if deadline is within 5 microsecs
        // 如果deadlineNanos小于5纳秒,则为0,,否则取整为1毫秒
        // 这段操作是为了向上取整,转成毫秒
        long timeoutMillis = deadlineToDelayNanos(deadlineNanos + 995000L) / 1000000L;
        // 如果timeoutMillis大于0,就阻塞selector同样的时间
        // 这段是为了获取最近的延时任务
        return timeoutMillis <= 0 ? selector.selectNow() : selector.select(timeoutMillis);
    }
  • runAllTasks

    此方法会取出可执行的任务,并执行。

    上一节添加任务讲的addTask,放入的任务就是在这里被执行的。

  • processSelectedKeys

    此方法在有IO事件时才触发,这个下面细讲

小结一下,run的死循环中主要判断有无IO事件,有则处理,处理完IO事件,再处理队列中的任务。
如果没有IO事件,也没有待处理的任务,则阻塞等待。

唤醒线程

对应代码

// wakeup(inEventLoop);
// 类NioEventLoop
protected void wakeup(boolean inEventLoop) {
    // 不是当前EventLoop在执行的时候,才需要唤醒
    // nextWakeupNanos放入AWAKE是阻止不必要的唤醒
    if (!inEventLoop && nextWakeupNanos.getAndSet(AWAKE) != AWAKE) {
        selector.wakeup();
    }
}

阻塞是selecto.select,那么唤醒就是selector.wakeup

王谷雨

一个苟且偷生的java程序员

文章评论