[Java知识库]Netty源码分析4:NioEventLoopGroup

NioEventLoopGroup及其父类和实现的接口作用分析

• NioEventLoopGroup内部维护了一组NioEventLoop，它们之间的关系有点类似于线程池和线程之间的关系，其类图如下：其中Iterable和ScheduledExecutorService属于jdk提供，
  

• EventExecutorGroup接口作为Netty定义事件执行器组的顶级接口，主要定义了优雅关闭EventExecutorGroup的方法。其次定义了一个next()方法用于轮询内部的EventExecutor。

• AbstractEventExecutorGroup初步实现了EventExecutorGroup接口，实际上内部是通过调用next()方法获取EventExecutor，然后执行EventExecutor对应的方法，简单的装饰者模式实现，查看EventExecutor接口发现其继承自EventExecutorGroup接口。

• MultithreadEventExecutorGroup定义了多线程事件执行器，即内部维护了一个EventExecutor[] children属性，当向EventExecutorGroup中提交任务时，实际上时通过next()方法轮询children的元素获取一个EventExecutor，最终通过EventExecutor来提交任务，此外Promise<?> terminationFuture属性用于记录MultithreadEventExecutorGroup的异步结果。Promise属于Netty封装Future的高级接口，提供了设置异步结果、完成状态、成功状态、异常结果的方法。此外还定义了一个EventExecutorChooserFactory.EventExecutorChooser chooser属性，该属性提供了如何轮询children并返回一个EventExecutor，通常时顺序轮询children，对于children.length为2的n次幂时采用类似HashMap中与运算求数组下标的方式，否则采用求余计算

• MultithreadEventLoopGroup则初步实现了事件轮询，其register相关接口均通过轮询next()来实现。即当我们调用MultithreadEventLoopGroup.regiter(…)接口时，实际上是通过轮询调用children元素调用其register(…)接口的，这保证了注册操作均匀的分布到children的每个元素中。此外还定义了一个DEFAULT_EVENT_LOOP_THREADS静态变量，该值等于当前运行环境的处理器数量 * 2

• NioEventLoopGroup即实现了NIO的事件轮询组

  • setIoRatio(int ioRatio)：设置IO时间的比例，因为我们可以向NioEventLoopGroup提交外部任务，因此NioEventLoopGroup除了监听通道事件外，还会额外执行外部任务。实际上该方法角遍历内部children并调用其setIoRatio方法，因此实际上是为内部EventExecutor(children)设定比例，默认比例为50，即IO操作与外部任务事件比例相同
  • rebuildSelectors()：该方法用于处理jdk多路复用器Selector的bug，实际上和setIoRatio一样也是遍历children调用rebuildSelectors实现
  • newChild()：创建一个EventExecutor，即初始化children中的每一个元素

new NioEventLoopGroup()实例化过程分析

    public NioEventLoopGroup() {
        this(0);
    }

    public NioEventLoopGroup(int nThreads) {
        this(nThreads, (Executor) null);
    }
    
    public NioEventLoopGroup(ThreadFactory threadFactory) {
        this(0, threadFactory, SelectorProvider.provider());
    }

    public NioEventLoopGroup(int nThreads, ThreadFactory threadFactory) {
        this(nThreads, threadFactory, SelectorProvider.provider());
    }

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

    public NioEventLoopGroup(
            int nThreads, ThreadFactory threadFactory, final SelectorProvider selectorProvider) {
        this(nThreads, threadFactory, selectorProvider, DefaultSelectStrategyFactory.INSTANCE);
    }

    public NioEventLoopGroup(int nThreads, ThreadFactory threadFactory,
        final SelectorProvider selectorProvider, final SelectStrategyFactory selectStrategyFactory) {
        super(nThreads, threadFactory, selectorProvider, selectStrategyFactory, RejectedExecutionHandlers.reject());
    }

从上面源码中可以看到无参构造器进行了一系列的构造方法重载，并调用了父类的构造器继续重载，其默认参数如下：

• int nThreads=0:默认创建0个EventExecutor，实际上使用的是父类中的DEFAULT_EVENT_LOOP_THREADS值=处理器核数 * 2
• Executor executor=null,默认执行器为null，实际上使用的ThreadPerTaskExecutor作为默认执行器，当向NioEventLoopGroup中提交任务时，实际上时提交到内部的一个EventExecutor中，而EventExecutor内部又维护了一个Executor，因此任务最终提交到了Executor中。
• SelectorProvider selectorProvider=SelectorProvider.provider():多路复用器Selector的提供商，直接使用jdk的提供商
• SelectStrategyFactory selectStrategyFactory：选择策略工厂用于创建选择策略，前面有说过EventExecutor在监听channel事件时也会执行外部提交的任务，通过选择策略来决定是否要进行阻塞监听channel事件还是去执行外部任务
• RejectedExecutionHandler reject:拒绝执行处理器，当EventExecutor使用有界队列来暂存外部提交的任务时，在队列已满的情况下将如何拒绝外部任务，默认实现时向外部抛出RejectedExecutionException
• 最终调用了父类MultithreadEventLoopGroup的构造器方法如下：

    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);
        }
    }
    protected MultithreadEventLoopGroup(int nThreads, Executor executor, Object... args) {
        super(nThreads == 0 ? DEFAULT_EVENT_LOOP_THREADS : nThreads, executor, args);
    }

MultithreadEventLoopGroup中的构造器重在方法在默认情况下将使用DEFAULT_EVENT_LOOP_THREADS作为nThreads 参数，DEFAULT_EVENT_LOOP_THREADS=构造器数量 * 2；也可以通过io.netty.eventLoopThreads来配置，最后又调用了父类MultithreadEventExecutorGroup的构造器方法如下：

protected MultithreadEventExecutorGroup(int nThreads, Executor executor, Object... args) {
	//使用默认的执行器选择工厂DefaultEventExecutorChooserFactory.INSTANCE
    this(nThreads, executor, DefaultEventExecutorChooserFactory.INSTANCE, args);
}
protected MultithreadEventExecutorGroup(int nThreads, Executor executor,
                                            EventExecutorChooserFactory chooserFactory, Object... args) {
        checkPositive(nThreads, "nThreads");
		//如果没有自定义执行器(该执行器最终被赋值给EventExecutor的成员变量)，则使用ThreadPerTaskExecutor
        if (executor == null) {
            executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());
        }
		//实例化children
        children = new EventExecutor[nThreads];
		//for循环将实例化children中的每一个元素
        for (int i = 0; i < nThreads; i ++) {
            boolean success = false;
            try {
            	//通过子类中的newChild()来实现
                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;
                        }
                    }
                }
            }
        }
		//实例化事件轮询器,即上述的默认的执行器选择工厂DefaultEventExecutorChooserFactory.INSTANCE
        chooser = chooserFactory.newChooser(children);
		//定义异步事件通知，该通知将被添加到事件执行器EventExecutor上，
		//其逻辑也是简单的当children的最后一个元素被成功初始化后设置当前Group的实例化结果
        final FutureListener<Object> terminationListener = new FutureListener<Object>() {
            @Override
            public void operationComplete(Future<Object> future) throws Exception {
                if (terminatedChildren.incrementAndGet() == children.length) {
                    terminationFuture.setSuccess(null);
                }
            }
        };
		//将上述通知添加到children中的每一个元素上
        for (EventExecutor e: children) {
            e.terminationFuture().addListener(terminationListener);
        }
		//构建一个不可更改的readonlyChildren用于遍历。
        Set<EventExecutor> childrenSet = new LinkedHashSet<EventExecutor>(children.length);
        Collections.addAll(childrenSet, children);
        readonlyChildren = Collections.unmodifiableSet(childrenSet);
}

• SelectStrategyFactory=DefaultSelectStrategyFactory.INSTANCE：select策略工厂用于决定当前是执行通道监听还是执额外任务

public final class DefaultSelectStrategyFactory implements SelectStrategyFactory {
    public static final SelectStrategyFactory INSTANCE = new DefaultSelectStrategyFactory();

    private DefaultSelectStrategyFactory() { }

    @Override
    public SelectStrategy newSelectStrategy() {
        return DefaultSelectStrategy.INSTANCE;
    }
}

final class DefaultSelectStrategy implements SelectStrategy {
    static final SelectStrategy INSTANCE = new DefaultSelectStrategy();

    private DefaultSelectStrategy() { }

	//计算策略：如果当前有任务，则由selectSupplier返回值决定，否则阻塞监听通道事件
	//selectSupplier不同的EventExecutor提供了不同实现，但均是非阻塞式的获取当前channel上的已准备好的事件的数量
	//包括java.nio.Selector.selectNow()，window下的多路复用器kqueueWaitNow()和linux下的多路复用器epollWaitNow()
    @Override
    public int calculateStrategy(IntSupplier selectSupplier, boolean hasTasks) throws Exception {
        return hasTasks ? selectSupplier.get() : SelectStrategy.SELECT;
    }
}

• chooserFactory = DefaultEventExecutorChooserFactory.INSTANCE事件执行器轮询工厂

public final class DefaultEventExecutorChooserFactory implements EventExecutorChooserFactory {

    public static final DefaultEventExecutorChooserFactory INSTANCE = new DefaultEventExecutorChooserFactory();

    private DefaultEventExecutorChooserFactory() { }

    @Override
    public EventExecutorChooser newChooser(EventExecutor[] executors) {
    	//如果nThreads是2的n次幂时，使用PowerOfTwoEventExecutorChooser选择器
    	//否则使用GenericEventExecutorChooser选择器
        if (isPowerOfTwo(executors.length)) {
            return new PowerOfTwoEventExecutorChooser(executors);
        } else {
            return new GenericEventExecutorChooser(executors);
        }
    }

    private static boolean isPowerOfTwo(int val) {
        return (val & -val) == val;
    }

    private static final class PowerOfTwoEventExecutorChooser implements EventExecutorChooser {
        private final AtomicInteger idx = new AtomicInteger();
        private final EventExecutor[] executors;

        PowerOfTwoEventExecutorChooser(EventExecutor[] executors) {
            this.executors = executors;
        }
		//2的n次幂通过每次索引+1后与运算获取对应索引位置的EventExecutor
        @Override
        public EventExecutor next() {
            return executors[idx.getAndIncrement() & executors.length - 1];
        }
    }

    private static final class GenericEventExecutorChooser implements EventExecutorChooser {
		//AtomicLong而不是AtomicInteger防止溢出
        private final AtomicLong idx = new AtomicLong();
        private final EventExecutor[] executors;

        GenericEventExecutorChooser(EventExecutor[] executors) {
            this.executors = executors;
        }

		//通过求余运算计算数组下标位置
        @Override
        public EventExecutor next() {
            return executors[(int) Math.abs(idx.getAndIncrement() % executors.length)];
        }
    }
}

• executor = new ThreadPerTaskExecutor(newDefaultThreadFactory())：EventExecutorGroup将提交的任务交给EventExecutor,而EventExecutor由将任务交给了ThreadPerTaskExecutor，ThreadPerTaskExecutor通过newDefaultThreadFactory()来创建一个线程用于执行最终的任务

public final class ThreadPerTaskExecutor implements Executor {
    private final ThreadFactory threadFactory;

    public ThreadPerTaskExecutor(ThreadFactory threadFactory) {
        this.threadFactory = ObjectUtil.checkNotNull(threadFactory, "threadFactory");
    }
	//ThreadPerTaskExecutor又将任务交给了threadFactory.newThread()方法来执行
    @Override
    public void execute(Runnable command) {
        threadFactory.newThread(command).start();
    }
}

DefaultThreadFactory创建线程的源码如下,忽略掉部分源码内容：

public class DefaultThreadFactory implements ThreadFactory {
	//没创建一个DefaultThreadFactory该poolId自增1,用于创建不同的线程组名
    private static final AtomicInteger poolId = new AtomicInteger();
	//没创建一个线程组中的线程该值自增1，用于创建同一线程组的不同的线程名
    private final AtomicInteger nextId = new AtomicInteger();
    private final String prefix;//线程名称的前缀：NioEventLoopGroup-1-
    private final boolean daemon;//线程是否为守护线程，默认false
    private final int priority;//线程优先级：1-10,使用默认值5
    protected final ThreadGroup threadGroup;/父线程组，默认null
    
    public DefaultThreadFactory(String poolName, boolean daemon, int priority, ThreadGroup threadGroup) {
        ObjectUtil.checkNotNull(poolName, "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;
    }
	
	//创建一个线程并设置相应属性
    @Override
    public Thread newThread(Runnable r) {
        Thread t = newThread(FastThreadLocalRunnable.wrap(r), prefix + nextId.incrementAndGet());
        try {
            if (t.isDaemon() != daemon) {
                t.setDaemon(daemon);
            }

            if (t.getPriority() != priority) {
                t.setPriority(priority);
            }
        } catch (Exception ignored) {
            // Doesn't matter even if failed to set.
        }
        return t;
    }
	
	//最终创建FastThreadLocalThread并返回，FastThreadLocalThread是Thread的子类
	//主要对jdk的ThreadLocal进行了优化，增加了访问速度
    protected Thread newThread(Runnable r, String name) {
        return new FastThreadLocalThread(threadGroup, r, name);
    }
}