Netty源码分析(一)——线程组初始化

继上篇文章讲述了Netty的基本使用之后，下面从源码角度分析Netty的执行流程过程中做了哪些事情。

我们知道，Netty底层是对JDK的NIO进行封装，并且其架构基础是基于主从Reactor模型设计的，主线程组负责处理Accept连接，从线程组负责处理IO事件，由于Netty处理IO的事件都是异步的，无法立即得到返回结果，它需要对事件添加监听器，当事件处理完毕时，调用isSuccess()来判断结果是否成功，或者调用sync()方法来使异步变为同步。

由于Netty底层流程涉及非常多的复杂流程，这篇文章就从Netty的启动入口——事件循环组的初始化来分析。

下面这段代码，定义两个事件循环组，分别为bossGroup(它用来接收Accept请求)和workerGroup(处理IO事件)。

// bossGroup处理客户端连接，绑定线程数1
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
// 处理IO事件的事件循环组，线程数为cpu的核心数*2
EventLoopGroup workerGroup = new NioEventLoopGroup();

NioEventLoopGroup继承关系：

NioEventLoop继承关系：

一个NioEventLoopGroup里面维护多个NioEventLoop，而每个NioEventLoop里面维护一个线程，事件的IO都是NioEventLoop里面对应的线程来执行的。

下面从源码方向分析：

初始化NioEventLoopGroup类：

public class NioEventLoopGroup extends MultithreadEventLoopGroup {
    // 无参构造方法
    public NioEventLoopGroup() {
        this(0);
    }
    
    // 指定线程数量的构造方法
    public NioEventLoopGroup(int nThreads) {
        this(nThreads, (Executor) null);
    }
    // 指定线程数量和执行器的构造方法
    public NioEventLoopGroup(int nThreads, Executor executor) {
        // SelectorProvider.provider()根据系统选择适配的选择器提供者
        this(nThreads, executor, SelectorProvider.provider());
    }
    
    //指定线程数量、执行器、Selector的构造方法
    public NioEventLoopGroup(
            int nThreads, Executor executor, final SelectorProvider selectorProvider) {
        this(nThreads, executor, selectorProvider, DefaultSelectStrategyFactory.INSTANCE);
    }
    
    //指定线程数量、执行器、Selector、SelectStrategyFactory的构造方法
    public NioEventLoopGroup(int nThreads, Executor executor, final SelectorProvider selectorProvider,
                             final SelectStrategyFactory selectStrategyFactory) {
        // 调用父类MultithreadEventLoopGroup的构造方法
        super(nThreads, executor, selectorProvider, selectStrategyFactory, RejectedExecutionHandlers.reject());
    }
}

===》父类MultithreadEventLoopGroup的构造方法

public abstract class MultithreadEventLoopGroup extends MultithreadEventExecutorGroup implements EventLoopGroup {
    private static final int DEFAULT_EVENT_LOOP_THREADS;

    static {
        // 线程数 = CPU的核数*2
        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) {
        // 如果初始化的时候没有赋值线程数，将会使用系统默认的线程数=CPU的核数*2
        // 然后调用父类MultithreadEventExecutorGroup的构造方法
        super(nThreads == 0 ? DEFAULT_EVENT_LOOP_THREADS : nThreads, executor, args);
    }
}

===》父类MultithreadEventExecutorGroup的构造方法

public abstract class MultithreadEventExecutorGroup extends AbstractEventExecutorGroup {
    private final EventExecutor[] children;
    
    protected MultithreadEventExecutorGroup(int nThreads, Executor executor, Object... args) {
        this(nThreads, executor, DefaultEventExecutorChooserFactory.INSTANCE, args);
    }
    
    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) {
            // 使用默认的ThreadPerTaskExecutor线程执行，它实现JDK Executor接口，是一个管理线程池的工具类
            // 它跟Executor一样，只不过自己定义了线程工厂
            executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());
        }

        // 初始化EventExecutor数组
        children = new EventExecutor[nThreads];

        for (int i = 0; i < nThreads; i ++) {
            boolean success = false;
            try {
                // 初始化EventExecutor对象，args表示多个参数
                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) {
                    // 如果初始化EventExecutor对象失败，则关闭所有已经创建的EventExecutor对象
                    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;
                        }
                    }
                }
            }
        }
        // 选择轮询执行器，这里面有两层判断：
        //（1）如果传进来的数是2的幂次方，则new PowerOfTwoEventExecutorChooser(executors)
        // (2) 如果不是，则是new GenericEventExecutorChooser(executors)
        // 最终，归根结底，都是实现轮询线程的效果
        chooser = chooserFactory.newChooser(children);

        //监听每个EventExecutor的关闭状态，所有EventExecutor都关闭了，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);
                }
            }
        };

        // 给每个EventExecutor添加关闭状态监听器
        for (EventExecutor e: children) {
            e.terminationFuture().addListener(terminationListener);
        }

        Set<EventExecutor> childrenSet = new LinkedHashSet<EventExecutor>(children.length);
        Collections.addAll(childrenSet, children);
        // 创建一个只读的集合
        readonlyChildren = Collections.unmodifiableSet(childrenSet);
    }
}

关键的核心：children[i] = newChild(executor, args); 它是给EventExecutor数组里面赋值每个NioEventLoop，且每个NioEventLoop关联一个SingleThreadEventExecutor

public class NioEventLoopGroup{
		@Override
    protected EventLoop newChild(Executor executor, Object... args) throws Exception {
        EventLoopTaskQueueFactory queueFactory = args.length == 4 ? (EventLoopTaskQueueFactory) args[3] : null;
        return new NioEventLoop(this, executor, (SelectorProvider) args[0],
            ((SelectStrategyFactory) args[1]).newSelectStrategy(), (RejectedExecutionHandler) args[2], queueFactory);
    }
}

至此，线程池组就初始化完毕。

NioEventLoop和NioEventLoopGroup的区别

1. NioEventLoopGroup是NioEventLoop的容器，提供了生成NioEventLoop的方法
2. NioEventLoop负责IO的读写之外,还兼顾处理以下两类任务：系统Task、定时任务

后续文章：ServerBootstrap的加载过程源码分析