Netty源码解析之线程池的实现（二）：创建线程与执行任务

前言

先看下面的代码：

public class MyTest {public static void main(String[] args) {//创建NioEventLoopGroupNioEventLoopGroup loopGroup = new NioEventLoopGroup(3);System.out.println(Thread.currentThread()+"准备执行任务");//执行任务for (int i = 0 ;i < 3 ; i++){loopGroup.execute(new Runnable() {@Overridepublic void run() {System.out.println(Thread.currentThread()+"执行任务");}});}//优雅的关闭loopGroup.shutdownGracefully();}
}

执行结果为：

由此我们可以看出，我们调用EventLoopGroup的execute(Runnable command)方法后，EventLoopGroup会分别创建不同的新线程来执行我们的任务。
那么问题来了，EventLoopGroup在什么时候创建的线程？如何创建的新线程？这个是本文需要论述的内容。

NioEventLoopGroup的实例化过程

NioEventLoopGroup的实例化过程做了大量的事情，现在以NioEventLoopGroup的无参构造为例。
首先我们如果执行了下面的代码，创建NioEventLoopGroup对象。

NioEventLoopGroup group = new NioEventLoopGroup();

1、NioEventLoopGroup中的构造函数调用

进入NioEventLoopGroup的源码中，看构造方法的调用链：

    //第一步，继续调用其他构造方法，并且传入线程数量为0public NioEventLoopGroup() {this(0);}//第二步：继续调用其他构造方法，增加参数Executor为nullpublic NioEventLoopGroup(int nThreads) {this(nThreads, (Executor) null);}//第三步，继续调用其他构造方法，增加参数与操作系统相关的SelectorProvider//这个SelectorProvider用来以后创建NIO的Selectorpublic NioEventLoopGroup(int nThreads, Executor executor) {this(nThreads, executor, SelectorProvider.provider());}//第三步：继续调用其他构造方法，增加参数SelectStrategyFactorypublic NioEventLoopGroup(int nThreads, Executor executor, final SelectorProvider selectorProvider) {this(nThreads, executor, selectorProvider, DefaultSelectStrategyFactory.INSTANCE);}//第四步：调用父类的构造方法，增加参数RejectedExecutionHandlerspublic NioEventLoopGroup(int nThreads, Executor executor, final SelectorProvider selectorProvider,final SelectStrategyFactory selectStrategyFactory) {super(nThreads, executor, selectorProvider, selectStrategyFactory, RejectedExecutionHandlers.reject());}

通过以上可以得知，如果使用NioEventLoopGroup的无参构造创建对象，最终会调用父类的构造方法，并且传入的参数中线程数量nThreads=0，执行器executor=null。

2、MultithreadEventLoopGroup的构造函数

下面，进入NioEventLoopGroup的父类MultithreadEventLoopGroup的构造方法中。

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

发现其做了两件事情：
（1）判断传入的线程数量是否为0，如果为0的话，使用默认线程数DEFAULT_EVENT_LOOP_THREADS。
这个默认DEFAULT_EVENT_LOOP_THREADS由一个静态代码块来初始化。

static {DEFAULT_EVENT_LOOP_THREADS = Math.max(1, SystemPropertyUtil.getInt("io.netty.eventLoopThreads", NettyRuntime.availableProcessors() * 2));}}

从上面的代码可以看出，如果没有指定io.netty.eventLoopThreads的值，则执行“NettyRuntime.availableProcessors() * 2”代码。

    public static int availableProcessors() {return holder.availableProcessors();}synchronized int availableProcessors() {if (this.availableProcessors == 0) {final int availableProcessors =SystemPropertyUtil.getInt("io.netty.availableProcessors",//获取CPU核心数Runtime.getRuntime().availableProcessors());setAvailableProcessors(availableProcessors);}return this.availableProcessors;}

最终，如果也没有指定io.netty.availableProcessors的值，则会调用Runtime.getRuntime().availableProcessors()来获取CPU核心数。

综上，Netty的默认线程数量是CPU核心数的2倍。

（2）调用父类的MultithreadEventExecutorGroup的构造方法，传入默认线程数和Executor等参数。

3、MultithreadEventExecutorGroup中的构造函数调用

在MultithreadEventExecutorGroup中，首先调用了下面的构造方法。这时候，传入的nThreads是CPU核心数的2倍，executor为null。

protected MultithreadEventExecutorGroup(int nThreads, Executor executor, Object... args) {this(nThreads, executor, DefaultEventExecutorChooserFactory.INSTANCE, args);}

上面的构造方法做了两件事：
（1）使用 DefaultEventExecutorChooserFactory.INSTANCE创建执行器选择器工厂，该工厂将在后面用于创建一个执行器选择器。这个执行器选择器就是调用NioEventLoopGroup的next()方法时，来选择一个执行器。

（2）继续调用另一个构造方法。
下面，进入这个构造方法，这也是最重要的构造方法，仍然位于MultithreadEventExecutorGroup之中。

    protected MultithreadEventExecutorGroup(int nThreads, Executor executor,EventExecutorChooserFactory chooserFactory, Object... args) {//检查传入的线程数量nThreads不能小于或等于0checkPositive(nThreads, "nThreads");//1、实例化executor，这里默认创建的是ThreadPerTaskExecutorif (executor == null) {executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());}//2、创建数组，用于存储执行器，数组数量需要等于线程数量。//因为每个执行器只有一个线程，即单线程的执行器。//在构造函数执行结束后，此数组最终存储的会是NioEventLoop实例。children = new EventExecutor[nThreads];//3、循环创建执行器for (int i = 0; i < nThreads; i ++) {//用来标记当前执行器是否创建成功boolean success = false;try {//4、使用newChild方法创建执行器实例children[i] = newChild(executor, args);success = true;} catch (Exception e) {// TODO: Think about if this is a good exception typethrow 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;}}}}}//5、根据EventExecutor数组的长度（即NioEventLoop的数量）来创建一个执行器选择器//根据上文，这里的chooserFactory是DefaultEventExecutorChooserFactory实例chooser = chooserFactory.newChooser(children);//6、创建监听器，以便在每个EventExecutor结束时，收到通知final FutureListener<Object> terminationListener = new FutureListener<Object>() {@Overridepublic void operationComplete(Future<Object> future) throws Exception {//如果所有的EventExecutor（一般即NioEventLoop）都进行了结束的通知if (terminatedChildren.incrementAndGet() == children.length) {//EventLoopGroup进行结束通知terminationFuture.setSuccess(null);}}};/** 每个EventExecutor（一般即NioEventLoop）在任务执行结束（即run()方法运行结束）* 后，最终会执行terminationFuture.setSuccess(null);代码进行监听器的通知。* 代码位于SingleThreadEventExecutor类的doStartThread()方法* */for (EventExecutor e: children) {e.terminationFuture().addListener(terminationListener);}//7、将所有的执行器EventExecutor存入只读的副本，以便于能通过迭代器获取Set<EventExecutor> childrenSet = new LinkedHashSet<EventExecutor>(children.length);Collections.addAll(childrenSet, children);readonlyChildren = Collections.unmodifiableSet(childrenSet);}

重点关注这两点：
（1）NioEventLoopGroup的父类MultithreadEventExecutorGroup中包含一个属性children，是一个EventExecutor型的数组，用来存储NioEventLoopGroup包含的所有执行器。
（2）执行器组NioEventLoopGroup包含的执行器在父类MultithreadEventExecutorGroup的构造方法中使用newChild方法来依次创建的。
（3）newChild方法创建执行器时传入了一个非常重要的参数executor，executor是使用“new ThreadPerTaskExecutor(newDefaultThreadFactory())”代码创建的ThreadPerTaskExecutor实例。
这里先剧透一下，newChild方法创建的执行器就是NioEventLoop实例，也就是说children数组中存储的是NioEventLoop实例。需要知道其原因，则要看newChild方法的具体实现。

4、NioEventLoopGroup中newChild方法的实现

在MultithreadEventExecutorGroup中，newChild是一个抽象方法，由子类NioEventLoopGroup实现。

protected EventLoop newChild(Executor executor, Object... args) throws Exception {//1、对参数进行整理SelectorProvider selectorProvider = (SelectorProvider) args[0];/*省略与本文内容无关的代码*///2、使用NioEventLoop的构造方法创建NioEventLoopreturn new NioEventLoop(this, executor, selectorProvider,selectStrategyFactory.newSelectStrategy(),rejectedExecutionHandler, taskQueueFactory, tailTaskQueueFactory);}

上面的newChild方法中，第1步可以先不看，直接看第2步，使用NioEventLoop的构造方法创建NioEventLoop。
这里需要注意的是，此时传递的executor参数是上文中使用“executor = new ThreadPerTaskExecutor(newDefaultThreadFactory())”代码创建的ThreadPerTaskExecutor实例。

5、创建NioEventLoop实例

接上文，NioEventLoopGroup中newChild最后new了NioEventLoop实例。下面，我们来看NioEventLoop实例的创建过程。

    //第1步：调用NioEventLoop的构造方法NioEventLoop(NioEventLoopGroup parent, Executor executor, SelectorProvider selectorProvider,SelectStrategy strategy, RejectedExecutionHandler rejectedExecutionHandler,EventLoopTaskQueueFactory taskQueueFactory, EventLoopTaskQueueFactory tailTaskQueueFactory) {//调用父类的构造方法super(parent, executor, false, newTaskQueue(taskQueueFactory), newTaskQueue(tailTaskQueueFactory),rejectedExecutionHandler);//对属性进行赋值this.provider = ObjectUtil.checkNotNull(selectorProvider, "selectorProvider");this.selectStrategy = ObjectUtil.checkNotNull(strategy, "selectStrategy");final SelectorTuple selectorTuple = openSelector();this.selector = selectorTuple.selector;this.unwrappedSelector = selectorTuple.unwrappedSelector;}//第2步：调用父类SingleThreadEventLoop的构造方法protected SingleThreadEventLoop(EventLoopGroup parent, Executor executor,boolean addTaskWakesUp, Queue<Runnable> taskQueue, Queue<Runnable> tailTaskQueue,RejectedExecutionHandler rejectedExecutionHandler) {//调用父类的构造方法super(parent, executor, addTaskWakesUp, taskQueue, rejectedExecutionHandler);//tailTasks 赋值，这个tailTasks的作用暂时不用管tailTasks = ObjectUtil.checkNotNull(tailTaskQueue, "tailTaskQueue");}//第3步：调用父类SingleThreadEventExecutor的构造方法protected SingleThreadEventExecutor(EventExecutorGroup parent, Executor executor,boolean addTaskWakesUp, Queue<Runnable> taskQueue,RejectedExecutionHandler rejectedHandler) {//调用父类的构造方法super(parent);//一些属性的赋值//是否添加唤醒任务this.addTaskWakesUp = addTaskWakesUp;//新任务被拒绝之前的最大待处理任务数，即为任务队列的容量this.maxPendingTasks = DEFAULT_MAX_PENDING_EXECUTOR_TASKS;//执行器executor属性this.executor = ThreadExecutorMap.apply(executor, this);//任务队列this.taskQueue = ObjectUtil.checkNotNull(taskQueue, "taskQueue");//拒绝策略this.rejectedExecutionHandler = ObjectUtil.checkNotNull(rejectedHandler, "rejectedHandler");}

这里，我们重点看第3步，即NioEventLoop的父类SingleThreadEventExecutor的构造方法。这里面有下面的这行代码：

this.executor = ThreadExecutorMap.apply(executor, this);

在这一行代码中，executor是上文中使用“new ThreadPerTaskExecutor(newDefaultThreadFactory())”代码创建的ThreadPerTaskExecutor实例，this就是NioEventLoop自身。

因此，可以看出，NioEventLoop对象里面含有executor属性，从其父类SingleThreadEventExecutor中继承而来，这个属性是ThreadPerTaskExecutor实例，这是NioEventLoop中线程的创建与运行的重点。

EventLoopGroup和EventLoop中的execute(Runnable command)方法

1、EventLoopGroup

在本文开头，我们使用NioEventLoopGroup的execute(Runnable command)方法执行任务。在Executor接口中，execute是抽象方法，需要由子类实现。在NioEventLoopGroup体系中，由NioEventLoopGroup的父类AbstractEventExecutorGroup对execute进行了重写实现。

    public void execute(Runnable command) {//先使用next()方法，选出一个执行器（即一个NioEventLoop），然后调用//执行器的execute(Runnable command)next().execute(command);}

其中的next()方法作用是在EventLoopGroup管理的执行器中选择一个，即选择一个NioEventLoop实例。
选择好NioEventLoop实例后，调用NioEventLoop的execute来执行任务。

2、EventLoop

添加任务到队列

EventLoop的execute(Runnable command)方法是在其父类SingleThreadEventExecutor中进行的实现：
实现代码如下：

    public void execute(Runnable task) {execute0(task);}private void execute0(@Schedule Runnable task) {//task 任务不能为 nullObjectUtil.checkNotNull(task, "task");execute(task, !(task instanceof LazyRunnable) && wakesUpForTask(task));}

最终调用下面的方法：

/** 执行任务的方法，这里并不直接运行Runnable任务的run()方法，* 而是将任务添加到队列中,然后启动线程去执行* */private void execute(Runnable task, boolean immediate) {// 当前线程是不是执行器线程boolean inEventLoop = inEventLoop();// 重点1、将任务添加到待普通执行任务队列taskQueue中// 注意这里是可以被不同线程调用的，所以有并发冲突问题。// 因此任务队列taskQueue 必须是一个线程安全的队列，就是可以处理并发问题。addTask(task);//如果当前线程不是执行器的线程if (!inEventLoop) {//重点2、要调用 startThread方法开启执行器线程，//这个方法做了判断，只有当执行器状态是 ST_NOT_STARTED 才会开启执行器线程startThread();// 如果执行器状态已经 Shutdown 之后，就要拒绝任务。// 注意这里的状态是已经 Shutdown 之后，所以不包括开始 Shutdown 的状态。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();}}}// 是否唤醒可能阻塞的执行器线程//addTaskWakesUp属性为调用addTask(Runnable)添加任务时是否能唤醒线程//immediate为任务是否立即执行if (!addTaskWakesUp && immediate) {wakeup(inEventLoop);}}

可以看出，EventLoop执行execute(Runnable task)方法的第一步就是
执行addTask(task)将任务添加到队列中。
其最终会将任务添加到taskQueue队列之中。

    //添加任务protected void addTask(Runnable task) {ObjectUtil.checkNotNull(task, "task");if (!offerTask(task)) {reject(task);}}//将任务加入普通队列final boolean offerTask(Runnable task) {if (isShutdown()) {reject();}return taskQueue.offer(task);}

判断是否需要开启线程—startThread()方法

在将任务添加到队列之后，判断当前线程是不是本执行器EventLoop的执行线程，如果不是的话，调用startThread()来创建执行器的线程。
当我们第一次调用EventLoopGroup或EventLoop的execute(Runnable task)方法时，执行器处于未运行状态，此时一定会进入到doStartThread()方法中。

    //开启执行器的线程private void startThread() {// 只有执行器处于未运行状态，才需要开启运行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);}}}}}

在上面的startThread()方法中，再次检查执行器状态为未运行，然后通过CAS方式，将执行器变成运行状态。准备就绪之后，就进入了正式创建线程的doStartThread()方法。

开启线程（一）—doStartThread()方法

下面，我们来重点看下doStartThread()方法是如何开启一个新线程的。

    private void doStartThread() {//执行本方法之前，因为没有创建线程，所以执行器的线程一定是nullassert thread == null;//1、新建一个Runnable任务//2、使用调用executor属性的execute方法来执行这个Runnable任务executor.execute(new Runnable() {@Overridepublic void run() {// 获取当前线程，赋值给 thread，就是执行器线程//注意：此时新线程肯定已经创建完成，否则不会赋值给threadthread = Thread.currentThread();// 若线程需要中断则中断线程if (interrupted) {thread.interrupt();}boolean success = false;// 更新最近一次执行任务时间updateLastExecutionTime();try {// 这个方法由子类实现，// 一般情况下，这个方法里面利用死循环，// 来获取待执行任务队列 taskQueue 中的任务并运行SingleThreadEventExecutor.this.run();// 标记启动成功success = true;} catch (Throwable t) {logger.warn("Unexpected exception from an event executor: ", t);} finally {/*此处省略了大量与本文内容无关的代码*/}}

在doStartThread()方法中，其实只做了两个动作，
1、新建一个Runnable任务。
2、使用调用executor属性的execute方法来执行这个Runnable任务。
并且，在Runnable任务的第一步，就获取当前线程，赋值给执行器EventLoop的thread属性。因此，executor在执行到Runnable任务的代码块时，肯定已经创建了新线程。

开启线程（二）—EventLoop中executor属性的execute方法

首先，EventLoop的executor属性到底是什么？
这个在上文已经交代过，就是“new ThreadPerTaskExecutor(newDefaultThreadFactory())”代码创建的ThreadPerTaskExecutor实例。
因此，doStartThread()方法中的executor.execute(new Runnable() {……}）这行代码实际上就是调用ThreadPerTaskExecutor的execute方法。

接下来，我们来看看ThreadPerTaskExecutor的execute方法到底做了什么。

1、ThreadPerTaskExecutor的execute方法作用是创建线程

进入ThreadPerTaskExecutor的源码，其源码非常简单。只有一个threadFactory属性，和一个execute(Runnable command)方法。

public final class ThreadPerTaskExecutor implements Executor {//线程工厂private final ThreadFactory threadFactory;//构造方法public ThreadPerTaskExecutor(ThreadFactory threadFactory) {this.threadFactory = ObjectUtil.checkNotNull(threadFactory, "threadFactory");}@Overridepublic void execute(Runnable command) {threadFactory.newThread(command).start();}
}

可以发现，其execute(Runnable command)方法就是使用其threadFactory属性的newThread方法来创建一个新线程，然后使用start()方法启动线程。

2、线程工厂threadFactory

ThreadPerTaskExecutor的threadFactory在其构造方法中赋值，由于之前使用的是“new ThreadPerTaskExecutor(newDefaultThreadFactory())”代码创建的ThreadPerTaskExecutor实例，我们继续进入newDefaultThreadFactory()方法之中。

    protected ThreadFactory newDefaultThreadFactory() {return new DefaultThreadFactory(getClass());}

newDefaultThreadFactory()方法作用是new了一个DefaultThreadFactory对象，传入的参数是当前实例的Class对象。由于我们现在创建的是NioEventLoopGroup实例，因此getClass()方法获取到的就是NioEventLoopGroup的Class对象。

这里传入Class对象，主要是为了获取类名，用来拼接线程的名称。这也是为什么在本文的开头，执行任务的线程的名称会是nioEventLoopGroup-2-1、nioEventLoopGroup-2-2这种类型（见本文“前言”部分的代码运行结果）。至于线程名称的具体拼接方式，可以在DefaultThreadFactory的源码中看，本文不再详述。

3、newThread方法的具体内容

在DefaultThreadFactory源码在，其newThread方法就是创建线程，但是创建的线程是FastThreadLocalThread类型。
这个FastThreadLocalThread是Thread的子类，即也是线程类。其具体的作用后续再写文章描述，和Netty对jdk中的ThreadLocal进行的改造和优化有关。

    public Thread newThread(Runnable r) {//1、将Runnable任务封装成FastThreadLocalRunnable//2、拼接线程名称//3、调用同名的newThread方法Thread t = newThread(FastThreadLocalRunnable.wrap(r), prefix + nextId.incrementAndGet());/* 此处省略代码 */return t;}protected Thread newThread(Runnable r, String name) {return new FastThreadLocalThread(threadGroup, r, name);}

到这里，可以看出来ThreadPerTaskExecutor的execute方法实际上就是创建了一个新的线程（FastThreadLocalThread实例），并且把新线程Runnable任务交给新线程去执行。这也就是为什么在doStartThread()方法中，executor.execute(new Runnable() {……}）这行代码执行到Runnable对象的内部时，已经是一个新线程去执行了。

开启线程（三）—NioEventLoop的executor属性执行的Runnable

在知道了NioEventLoop在执行execute(Runnable command)方法时，是调用ThreadPerTaskExecutor实例（在NioEventLoop中是executor属性）的execute方法，先创建一个新线程（FastThreadLocalThread实例），然后由新线程去运行Runnable任务。

在doStartThread()方法中，我们可以看到Runnable任务的内容：

executor.execute(new Runnable() {@Overridepublic void run() {// 获取当前线程，赋值给 thread，就是执行器线程//注意：此时新线程肯定已经创建完成，否则不会赋值给threadthread = Thread.currentThread();/*省略与本文无关的代码*/boolean success = false;// 更新最近一次执行任务时间updateLastExecutionTime();try {// 这个方法由子类实现，// 一般情况下，这个方法里面利用死循环，// 来获取待执行任务队列 taskQueue 中的任务并运行SingleThreadEventExecutor.this.run();// 标记启动成功success = true;} catch (Throwable t) {logger.warn("Unexpected exception from an event executor: ", t);} finally {/*此处省略了大量与本文内容无关的代码*/}

在Runnable内部的run()方法中，并没有直接去执行NioEventLoop的execute(Runnable command)方法所提交的任务，因为上文说过，这个任务已经存入到NioEventLoop中的任务队列taskQueue之中。这里面最重要的就是下面的这行代码：

SingleThreadEventExecutor.this.run();

ThreadPerTaskExecutor创建的新线程执行了一个Runnable任务，在Runnable任务里面又运行了SingleThreadEventExecutor的run()方法。但是SingleThreadEventExecutor的run()方法是抽象方法，需要由子类去实现。在Netty中，一般情况下这个run()方法的功能之一就是循环从任务队列中获取任务并运行。
如在NioEventLoop中，run()方法要做的事情之一就是队列中任务的处理。

总结

1、调用NioEventLoopGroup的构造方法时，会创建一个ThreadPerTaskExecutor实例，然后将其赋值给每个NioEventLoop的executor属性。
1、调用NioEventLoopGroup的execute(Runnable command)方法时，会首先使用next()方法选择一个NioEventLoop，然后调用NioEventLoop的execute方法执行任务。
2、调用NioEventLoop的execute方法时，会先将任务加入到队列中，然后使用NioEventLoop中的executor（ThreadPerTaskExecutor的实例）来创建线程。
3、创建的新线程会执行NioEventLoop的run()方法，在NioEventLoop的run()方法中会执行任务队列中的任务。
虽然Netty源码比较复杂，但是一切还是有规可循的。