EventLoop 相关源码分析（三十）

清疚 2022-10-29 14:23 172阅读 0赞

今天进行EventLoop 相关源码分析：

1、EventLoop结构图如下

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70][]

**2、** **EventLoopGroup**  **和**  **EventLoop**  **代码分析**

/*线程组*/
    EventLoopGroup group  = new NioEventLoopGroup();

执行这行代码时会发生什么？，由 NioEventLoopGroup 开始，一路调用，到达 MultithreadEventLoopGroup，如果没有指定创建的线程数量，则默认创建的线程个数为

DEFAULT\_EVENT\_LOOP\_THREADS ，该数值为：处理器数量  x2 。

protected MultithreadEventLoopGroup(int nThreads, Executor executor, Object... args) {
            super(nThreads == 0 ? DEFAULT_EVENT_LOOP_THREADS : nThreads, executor, args);
        }

查看线程数：

private static final int DEFAULT_EVENT_LOOP_THREADS;
    
    static {
        DEFAULT_EVENT_LOOP_THREADS = Math.max(1, SystemPropertyUtil.getInt(
                "io.netty.eventLoopThreads", NettyRuntime.availableProcessors() * 2));
    
        if (logger.isDebugEnabled()) {
            logger.debug("-Dio.netty.eventLoopThreads: {}", DEFAULT_EVENT_LOOP_THREADS);
        }
    }

点击 super(nThreads == 0 ? DEFAULT\_EVENT\_LOOP\_THREADS : nThreads, executor, args);最终由 MultithreadEventExecutorGroup 实例化，在这个构造方法中，实例化了每个 EventLoop 所需要的执行器 Executor

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) {
                executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());
            }
    
            children = new EventExecutor[nThreads];
    
            for (int i = 0; i < nThreads; i ++) {
                boolean success = false;
                try {
                    children[i] = newChild(executor, args);
                    success = true;
                } catch (Exception e) {
                    // TODO: Think about if this is a good exception type
                    throw 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;
                            }
                        }
                    }
                }
            }
    
            chooser = chooserFactory.newChooser(children);
    
            final FutureListener<Object> terminationListener = new FutureListener<Object>() {
                @Override
                public 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);
        }

由此可见，每个 NioEventLoop 的执行器为 ThreadPerTaskExecutor，ThreadPerTaskExecutor 实现了 Executor 接口，并会在 execute 方法中启动真正的线程，但是要和 NioEventLoop 的线程挂钩则在 SingleThreadEventExecutor 的 doStartThread 方法里。

public final class ThreadPerTaskExecutor implements Executor {
        private final ThreadFactory threadFactory;
    
        public ThreadPerTaskExecutor(ThreadFactory threadFactory) {
            if (threadFactory == null) {
                throw new NullPointerException("threadFactory");
            }
            this.threadFactory = threadFactory;
        }
    
        @Override
        public void execute(Runnable command) {
            threadFactory.newThread(command).start();
        }
    }

接下来，new 出 EventExecutor（实际是 NioEventLoop）的实例数组，并在循环里 new 每个具体的 EventLoop 实例

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 1][]

那么在 NioEventLoop 实例的构造方法里又做了什么事情呢？

@Override
        protected EventLoop newChild(Executor executor, Object... args) throws Exception {
            return new NioEventLoop(this, executor, (SelectorProvider) args[0],
                ((SelectStrategyFactory) args[1]).newSelectStrategy(), (RejectedExecutionHandler) args[2]);
        }

作为 IO 事件处理的主要组件，内部持有了 Selector、SelectionKey 的集合，所以构造方法中执行了关键方法 openSelector()，最终通过 JDK 的 api 拿到 selector 的实例，作用和我们 通过原生 JDK 的 NIO 编程中创建选择器是一样的。

**3、原生NIO实现：**

/*创建选择器的实例*/
    selector = Selector.open();

继续：

public static Selector open() throws IOException {
        return SelectorProvider.provider().openSelector();
    }

继续：

public abstract AbstractSelector openSelector()
            throws IOException;

**4、Netty实现** **：**

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;
        }

点击：final SelectorTuple selectorTuple = openSelector();

![20210306202215964.png][]

点击：unwrappedSelector = provider.openSelector();

public abstract AbstractSelector openSelector()
            throws IOException;

同时，通过前面 NioEventLoop 类关系图 我们可以知道，EventLoop  本质上是一个线程 池，EventLoop  内部维护着一个线程  Thread  和几个阻塞队列，所以  EventLoop  可以看成只有 一个线程的线程池（SingleThreadPool ） 每个 EventLoop  包含的线程  Thread  定义在父类  SingleThreadEventExecutor  中，每个 EventLoop 包含两个队列， taskQueue  来自父类  SingleThreadEventExecutor ，保存各种任务， 比如处理事件等等，tailTask  来自父类  SingleThreadEventLoop ，用于每次事件循环后置任务处理

SingleThreadEventExecutor 核心代码：

public abstract class SingleThreadEventExecutor extends AbstractScheduledEventExecutor implements OrderedEventExecutor {
    
        static final int DEFAULT_MAX_PENDING_EXECUTOR_TASKS = Math.max(16,
                SystemPropertyUtil.getInt("io.netty.eventexecutor.maxPendingTasks", Integer.MAX_VALUE));
    
        private static final InternalLogger logger =
                InternalLoggerFactory.getInstance(SingleThreadEventExecutor.class);
    
        private static final int ST_NOT_STARTED = 1;
        private static final int ST_STARTED = 2;
        private static final int ST_SHUTTING_DOWN = 3;
        private static final int ST_SHUTDOWN = 4;
        private static final int ST_TERMINATED = 5;
    
        private static final Runnable WAKEUP_TASK = new Runnable() {
            @Override
            public void run() {
                // Do nothing.
            }
        };
        private static final Runnable NOOP_TASK = new Runnable() {
            @Override
            public void run() {
                // Do nothing.
            }
        };
    
        private static final AtomicIntegerFieldUpdater<SingleThreadEventExecutor> STATE_UPDATER =
                AtomicIntegerFieldUpdater.newUpdater(SingleThreadEventExecutor.class, "state");
        private static final AtomicReferenceFieldUpdater<SingleThreadEventExecutor, ThreadProperties> PROPERTIES_UPDATER =
                AtomicReferenceFieldUpdater.newUpdater(
                        SingleThreadEventExecutor.class, ThreadProperties.class, "threadProperties");
    
        private final Queue<Runnable> taskQueue;
    
        private volatile Thread thread;

SingleThreadEventLoop 核心代码：

public abstract class SingleThreadEventLoop extends SingleThreadEventExecutor implements EventLoop {
    
        protected static final int DEFAULT_MAX_PENDING_TASKS = Math.max(16,
                SystemPropertyUtil.getInt("io.netty.eventLoop.maxPendingTasks", Integer.MAX_VALUE));
    
        private final Queue<Runnable> tailTasks;

作为 IO 事件处理的主要组件，必然离不开对事件的处理机制，在 NioEventLoop 的 run 方法，就有 selector 上进行 select 和调用 processSelectedKeys()处理各种事件集。

核心代码：

@Override
        protected void run() {
            for (;;) {
                try {
                    switch (selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())) {
                        case SelectStrategy.CONTINUE:
                            continue;
                        case SelectStrategy.SELECT:
                            select(wakenUp.getAndSet(false));

点击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 (;;) {
                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 ++;

run方法中 processSelectedKeys();调用：

cancelledKeys = 0;
    needsToSelectAgain = false;
    final int ioRatio = this.ioRatio;
    if (ioRatio == 100) {
        try {
            processSelectedKeys();
        } finally {
            // Ensure we always run tasks.
            runAllTasks();
        }

继续：

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

继续：

private void processSelectedKeysPlain(Set<SelectionKey> selectedKeys) {
        // check if the set is empty and if so just return to not create garbage by
        // creating a new Iterator every time even if there is nothing to process.
        // See https://github.com/netty/netty/issues/597
        if (selectedKeys.isEmpty()) {
            return;
        }
    
        Iterator<SelectionKey> i = selectedKeys.iterator();
        for (;;) {
            final SelectionKey k = i.next();
            final Object a = k.attachment();
            i.remove();
    
            if (a instanceof AbstractNioChannel) {
                processSelectedKey(k, (AbstractNioChannel) a);
            } else {
                @SuppressWarnings("unchecked")
                NioTask<SelectableChannel> task = (NioTask<SelectableChannel>) a;
                processSelectedKey(k, task);
            }
    
            if (!i.hasNext()) {
                break;
            }
    
            if (needsToSelectAgain) {
                selectAgain();
                selectedKeys = selector.selectedKeys();
    
                // Create the iterator again to avoid ConcurrentModificationException
                if (selectedKeys.isEmpty()) {
                    break;
                } else {
                    i = selectedKeys.iterator();
                }
            }
        }
    }

继续：

private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
        final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();
        if (!k.isValid()) {
            final EventLoop eventLoop;
            try {
                eventLoop = ch.eventLoop();
            } catch (Throwable ignored) {
                // If the channel implementation throws an exception because there is no event loop, we ignore this
                // because we are only trying to determine if ch is registered to this event loop and thus has authority
                // to close ch.
                return;
            }
            // Only close ch if ch is still registered to this EventLoop. ch could have deregistered from the event loop
            // and thus the SelectionKey could be cancelled as part of the deregistration process, but the channel is
            // still healthy and should not be closed.
            // See https://github.com/netty/netty/issues/5125
            if (eventLoop != this || eventLoop == null) {
                return;
            }
            // close the channel if the key is not valid anymore
            unsafe.close(unsafe.voidPromise());
            return;
        }
    
        try {
            int readyOps = k.readyOps();
            // We first need to call finishConnect() before try to trigger a read(...) or write(...) as otherwise
            // the NIO JDK channel implementation may throw a NotYetConnectedException.
            if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
                // remove OP_CONNECT as otherwise Selector.select(..) will always return without blocking
                // See https://github.com/netty/netty/issues/924
                int ops = k.interestOps();
                ops &= ~SelectionKey.OP_CONNECT;
                k.interestOps(ops);
    
                unsafe.finishConnect();
            }
    
            // Process OP_WRITE first as we may be able to write some queued buffers and so free memory.
            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();
            }
    
            // Also check for readOps of 0 to workaround possible JDK bug which may otherwise lead
            // to a spin loop
            if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
                unsafe.read();
            }
        } catch (CancelledKeyException ignored) {
            unsafe.close(unsafe.voidPromise());
        }
    }

5、  **NioEventLoopGroup**  **类关系图**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 2][]

到此，EventLoop 相关源码分析完成，下篇分析Netty服务端启动源码，敬请期待。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70]: /images/20221024/266398aa825a42fd832a27c9bbd0ba8c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 1]: /images/20221024/e653e4bcafc34381a7b12afc68ab62c2.png
[20210306202215964.png]: /images/20221024/0c2b4c067aa74f4996bdedc924c2b17f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 2]: /images/20221024/175498017b04426ab4f2885f0f6684fd.png