netty源码阅读之NioEventLoop之NioEventLoop创建

喜欢ヅ旅行 2022-05-17 08:58 143阅读 0赞

从new NioEventLoopGroup()进入分析NioEventLoop创建，NioEventLoopGroup创建NioEventLoop分为以下几个过程：

1、创建线程创建(执行)器：new ThreadPerTaskExecutor()

2、构造NioEventLoop：for\{newChild()\}

3、创建线程选择器：chooserFactory.newChooser()

从new NioEventLoopGroup()进入，一层层进入，会有下面一段代码：

protected MultithreadEventExecutorGroup(int nThreads, Executor executor,
                                                EventExecutorChooserFactory chooserFactory, Object... args) {
            ...
            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 {
                    ...
            }
    
            chooser = chooserFactory.newChooser(children);
            ...
        }

这就是刚刚说的三个过程，我们一步步分析

### 一、创建线程创建器：new ThreadPerTaskExecutor() ###

1、每次执行任务都会创建一个线程实体

2、NioEventLoop线程的命名规则nioEventLoop-(第几个线程池)-(这个线程池的第几个线程)

从new NioEventLoopGroup()进入，一层层进入，看到如下：

public NioEventLoopGroup(int nThreads, Executor executor) {
            this(nThreads, executor, SelectorProvider.provider());
        }

也就是，每个NioEventLoopGroup都会有一个selector，从这里创建。

继续进去，有如下代码：

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

如果没有定义线程数量，也就是为0的时候，就使用DEFAULT\_EVENT\_LOOP\_THREADS，它的定义为：

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

默认为系统线程数的两倍。

然后，继续进入，来到我们最开始分析的：

if (executor == null) {
                executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());
            }

查看这个new ThreadPerTaskExecutor()的定义

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

就是传进去一个ThreadFacotory，通过这个threadFactory产生线程。

回答我们之前的一个问题：NioEventLoop什么时候创建线程，在执行ThreadPerTaskExecutor这个execute方法的时候，把一个Runnable传进去创建线程。也就是每次执行任务的时候，创建一个线程实体。

线程执行器的作用：通过一个ThreadFactory，每次执行任务都会构造一个线程，然后去运行。（**线程执行器执行execute方法的时候，会构建一个线程（这个线程执行的时候也就是运行run的时候会执行NioEventLoop的run方法）；而NioEventLoop执行execute的时候，就是把任务放到任务队列里面**）

回到newDefaultThreadFactory()，查看实现，可以知道上面一步的threadFactory就是DefaultThreadFactory，在里面，有一个

public DefaultThreadFactory(Class<?> poolType, boolean daemon, int priority) {
            this(toPoolName(poolType), daemon, priority);
        }

前面通过getClass()得到poolType其实就是xx.xx.xx.NioEventLoop

toPoolName(poolType),他的实现就是：

public static String toPoolName(Class<?> poolType) {
            if (poolType == null) {
                throw new NullPointerException("poolType");
            }
    
            String poolName = StringUtil.simpleClassName(poolType);
            switch (poolName.length()) {
                case 0:
                    return "unknown";
                case 1:
                    return poolName.toLowerCase(Locale.US);
                default:
                    if (Character.isUpperCase(poolName.charAt(0)) && Character.isLowerCase(poolName.charAt(1))) {
                        return Character.toLowerCase(poolName.charAt(0)) + poolName.substring(1);
                    } else {
                        return poolName;
                    }
            }
        }

返回其实就是nioEventLoop，因为poolType是NioEventLoop。

一层层点，查看到另外一个构造函数：

public DefaultThreadFactory(String poolName, boolean daemon, int priority, ThreadGroup threadGroup) {
            if (poolName == null) {
                throw new NullPointerException("poolName");
            }
            if (priority < Thread.MIN_PRIORITY || priority > Thread.MAX_PRIORITY) {
                throw new IllegalArgumentException(
                        "priority: " + priority + " (expected: Thread.MIN_PRIORITY <= priority <= Thread.MAX_PRIORITY)");
            }
    
            prefix = poolName + '-' + poolId.incrementAndGet() + '-';
            this.daemon = daemon;
            this.priority = priority;
            this.threadGroup = threadGroup;
        }

添加了两个连接符，并把当前线程池的id获取到并且加一了。所以在这个类实现newThread这里,线程的名称就出来了：

@Override
        public Thread newThread(Runnable r) {
            Thread t = newThread(new DefaultRunnableDecorator(r), prefix + nextId.incrementAndGet());
            ...
            return t;
        }

至于netty它底层的线程，就是这个自定义的FastThreadLocalThread：

protected Thread newThread(Runnable r, String name) {
            return new FastThreadLocalThread(threadGroup, r, name);
        }

### 二、构造NioEventLoop：for\{newChild()\} ###

这一步做了三件事情：

1、保存上面创建的线程执行器ThreadPerTaskExecutor

2、创建一个MpscQueue（任务队列）

3、创建一个selector（轮询注册到它上面的连接）

首先我们看一个类图：

![70][]

newChild出来的就是NioEventLoop，它继承自SingleThreadEventExecutor，点击newChild的NioEventLoop的实现，然后一直点，super构造方法的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);
            rejectedExecutionHandler = ObjectUtil.checkNotNull(rejectedHandler, "rejectedHandler");
        }

在这里，它把刚刚传经来的executor绑定进去了，后面创建NioEventLoop底层线程的时候需要用到。

然后，newTaskQueue创建的是mpscQueue，这个队列的作用是当外部线程执行netty的一些任务的时候，如果判断到不是当前NioEventLoop对应的线程去执行的时候，就把它放到队列里面，然后由NioEventLoop对应的线程去执行：

@Override
        protected Queue<Runnable> newTaskQueue(int maxPendingTasks) {
            // This event loop never calls takeTask()
            return PlatformDependent.newMpscQueue(maxPendingTasks);
        }

看看定义：

/**
         * Create a new {@link Queue} which is safe to use for multiple producers (different threads) and a single
         * consumer (one thread!).
         */
        public static <T> Queue<T> newMpscQueue(final int maxCapacity) {
            return Mpsc.newMpscQueue(maxCapacity);
        }

也就是，一个消费者多个生产者，也就是有多个外部线程，一个NioEventLoop去消费。

另外创造selector是在这里实现的：

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;
            selector = openSelector();
            selectStrategy = strategy;
        }

### 三、创建线程选择器：chooserFactory.newChooser() ###

这里面对线程的轮询采用了优化的方式

isPowerOfTwo()判断是否是2的幂

1、是

采用PowerOfTwoEventExecutorChooser（优化点）,轮询的方式：

index++&(lenght-1)

2、不是

GenericEventExecutorChooser(),轮询的方式：

abs(index++%length)

直接贴代码看好了：

public final class DefaultEventExecutorChooserFactory implements EventExecutorChooserFactory {
    
        public static final DefaultEventExecutorChooserFactory INSTANCE = new DefaultEventExecutorChooserFactory();
    
        private DefaultEventExecutorChooserFactory() { }
    
        @SuppressWarnings("unchecked")
        @Override
        public EventExecutorChooser newChooser(EventExecutor[] executors) {
            if (isPowerOfTwo(executors.length)) {
                return new PowerOfTowEventExecutorChooser(executors);
            } else {
                return new GenericEventExecutorChooser(executors);
            }
        }
    
        private static boolean isPowerOfTwo(int val) {
            return (val & -val) == val;
        }
    
        private static final class PowerOfTowEventExecutorChooser implements EventExecutorChooser {
            private final AtomicInteger idx = new AtomicInteger();
            private final EventExecutor[] executors;
    
            PowerOfTowEventExecutorChooser(EventExecutor[] executors) {
                this.executors = executors;
            }
    
            @Override
            public 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;
            }
    
            @Override
            public EventExecutor next() {
                return executors[Math.abs(idx.getAndIncrement() % executors.length)];
            }
        }
    }

解释更优的原因：计算机底层，&比%计算速度更快，&直接通过二进制操作可以实现，但是%计算机底层没有二进制简单的实现，需要通过复杂实现。

正确的原因：实现目的就是轮询到最后一个，就开始重头轮询（循环取下标，也可以说是取模），普通方式的实现是可以的，我们现在解析PowerOfTwoEventExecutorChooser：

假设现在有16个线程，二进制为10000，length-1为1111，线程索引从0开始，idx也从0开始

1、轮询到第15次，也就是idx为1110，结果为1110，由于线程索引从0开始，就是第15个线程，正确

2、轮询到第16次，也就是idx为1111，结果为1111，由于线程索引从0开始，就是第16个线程，正确

3、关键是第17次，idx就是10000，结果为0000，由于线程索引从0开始，也就是第1个线程，正确

4、轮询到第18次，idx就是10001，结果为0001，由于线程索引从0开始，也就是第2个线程，正确

是不是很神奇？

因为2次幂的length减一，所有的位都是1。&的时候，在它前面的位置都会置0，也就是又可以重新开始了。

这一点，其实和hashMap1.7hash取余的方式很像。

[70]: /images/20220517/72f333df3f88455d855cfdcd5a40b279.png