Netty 源码解析（五）: Netty 的线程池分析

今天是猿灯塔“365篇原创计划”第五篇。 接下来的时间灯塔君持续更新Netty系列一共九篇

Netty 源码解析（一）: 开始

Netty 源码解析（二）: Netty 的 Channel

Netty 源码解析（三）: Netty的 Future 和 Promise

Netty 源码解析（四）: Netty 的 ChannelPipeline

当前：Netty 源码解析（五）: Netty 的线程池分析

Netty 源码解析（六）: Channel 的 register 操做

Netty 源码解析（七）: NioEventLoop 工做流程

Netty 源码解析（八）: 回到 Channel 的 register 操做

Netty 源码解析（九）: connect 过程和 bind 过程分析

今天呢！灯塔君跟你们讲：

Netty 的线程池分析

Netty 中的线程池 EventLoopGroup

接下来，咱们来分析 Netty 中的线程池。Netty 中的线程池比较很差理解，由于它的类比较多，并且它们之间的关系错综复杂。看下图，感觉下 NioEventLoop 类和 NioEventLoopGroup 类的继承结构：

这张图我按照继承关系整理而来，你们仔细看一下就会发现，涉及到的类确实挺多的。本节来给你们理理清楚这部份内容。

首先，咱们说的 Netty 的线程池，指的就是 NioEventLoopGroup 的实例；线程池中的单个线程，指的是右边 NioEventLoop 的实例。

回顾下咱们第一节介绍的 Echo 例子，客户端和服务端的启动代码中，最开始咱们老是先实例化 NioEventLoopGroup：

// EchoClient 代码最开始：

EventLoopGroup group = new NioEventLoopGroup();

// EchoServer 代码最开始：

EventLoopGroup bossGroup = new NioEventLoopGroup(1);

EventLoopGroup workerGroup = new NioEventLoopGroup();

下面，咱们就从 NioEventLoopGroup 的源码开始进行分析。 咱们打开 NioEventLoopGroup 的源码，能够看到，NioEventLoopGroup 有多个构造方法用于参数设置，最简单地，咱们采用无参构造函数，或仅仅设置线程数量就能够了，其余的参数采用默认值。

好比上面的代码中，咱们只在实例化 bossGroup 的时候指定了参数，表明该线程池须要一个线程。

publicNioEventLoopGroup(){
this(0);
}
publicNioEventLoopGroup(intnThreads){
this(nThreads,(Executor)null);
}
...
//参数最全的构造方法
publicNioEventLoopGroup(intnThreads,Executorexecutor,EventExecutorChooserFactorychooserFactory,
finalSelectorProviderselectorProvider,
finalSelectStrategyFactoryselectStrategyFactory,
finalRejectedExecutionHandlerrejectedExecutionHandler){
//调用父类的构造方法
super(nThreads,executor,chooserFactory,selectorProvider,selectStrategyFactory,rejectedExecutionHandler);
}
复制代码

咱们来稍微看一下构造方法中的各个参数：

• nThreads：这个最简单，就是线程池中的线程数，也就是 NioEventLoop 的实例数量。
• executor：咱们知道，咱们自己就是要构造一个线程池（Executor），为何这里传一个 executor 实例呢？它其实不是给线程池用的，而是给 NioEventLoop 用的，之后再说。
• chooserFactory：当咱们提交一个任务到线程池的时候，线程池须要选择（choose）其中的一个线程来执行这个任务，这个就是用来实现选择策略的。
• selectorProvider：这个简单，咱们须要经过它来实例化 JDK 的 Selector，能够看到每一个线程池都持有一个 selectorProvider 实例。
• selectStrategyFactory：这个涉及到的是线程池中线程的工做流程，在介绍 NioEventLoop 的时候会说。
• rejectedExecutionHandler：这个也是线程池的好朋友了，用于处理线程池中没有可用的线程来执行任务的状况。在 Netty 中稍微有一点点不同，这个是给 NioEventLoop 实例用的，之后咱们再详细介绍。

这里介绍这些参数是但愿你们有个印象而已，你们发现没有，在构造 NioEventLoopGroup 实例时的好几个参数，都是用来构造 NioEventLoop 用的。下面，咱们从 NioEventLoopGroup 的无参构造方法开始，跟着源码走：

publicNioEventLoopGroup(){
this(0);
}

publicNioEventLoopGroup(){
this(0);
}

复制代码

而后一步步走下去，到这个构造方法：

publicNioEventLoopGroup(intnThreads,ThreadFactorythreadFactory,finalSelectorProviderselectorProvider,finalSelectStrategyFactoryselectStrategyFactory){

super(nThreads,threadFactory,selectorProvider,selectStrategyFactory,RejectedExecutionHandlers.reject());
}复制代码

你们本身要去跟一下源码，这样才知道中间设置了哪些默认值，下面这几个参数都被设置了默认值：

• selectorProvider = SelectorProvider.provider() 这个没什么好说的，调用了 JDK 提供的方法
• selectStrategyFactory = DefaultSelectStrategyFactory.INSTANCE 这个涉及到的是线程在作 select 操做和执行任务过程当中的策略选择问题，在介绍 NioEventLoop 的时候会用到。
• rejectedExecutionHandler = RejectedExecutionHandlers.reject() 你们进去看一下 reject() 方法，也就是说，Netty 选择的默认拒绝策略是：抛出异常

跟着源码走，咱们会来到父类 MultithreadEventLoopGroup 的构造方法中：

protectedMultithreadEventLoopGroup(intnThreads,ThreadFactorythreadFactory,Object...args){
super(nThreads==0?DEFAULT_EVENT_LOOP_THREADS:nThreads,threadFactory,args);
}复制代码

这里咱们发现，若是采用无参构造函数，那么到这里的时候，默认地 nThreads 会被设置为 CPU 核心数 *2。你们能够看下 DEFAULT_EVENT_LOOP_THREADS 的默认值，以及 static 代码块的设值逻辑。咱们继续往下走：

protectedMultithreadEventExecutorGroup(intnThreads,ThreadFactorythreadFactory,Object...args){
this(nThreads,threadFactory==null?null:newThreadPerTaskExecutor(threadFactory),args);复制代码

到这一步的时候，new ThreadPerTaskExecutor(threadFactory) 会构造一个 executor。

咱们如今还不知道这个 executor 怎么用。这里咱们先看下它的源码： public final class ThreadPerTaskExecutor implements Executor { private final`ThreadFactorythreadFactory; public T hreadPerTaskExecutor(T hreadFactorythreadFactory) {` `i`f(`threadFactory==null){ th row` n`ewNul`l`PointerException("threadFactory"); } this .threa dFacto ry=threadFactory; } @Override` p`u`bl`ic void e`x`ecute(`R`unna`b`lecommand) { //为每一个任务新建一个线 程 threadFactory.n ewThread(command).start(); } } Executor 做为线程池的 最顶 层接`口， 咱们知道，它只有一个 execute(runnable) 方法，从上面咱们能够看到，实现类 ThreadPerTaskExecutor 的逻辑就是每来一个任务，新建一个线程。咱们先记住这个，前面也说了，它是给 NioEventLoop 用的，不是给 NioEventLoopGroup 用的。

.一步设置完了 executor，咱们继续往下看：

protectedMultithreadEventExecutorGroup(intnThreads,Executorexecutor,Object...args){
this(nThreads,executor,DefaultEventExecutorChooserFactory.INSTANCE,args);
}
这一步设置了 chooserFactory，用来实现从线程池中选择一个线程的选择策略。
复制代码

ChooserFactory 的逻辑比较简单，咱们看下 DefaultEventExecutorChooserFactory 的实现： Override public`EventExecutorChooserne wChooser(EventExecutor[]executors) { if (is PowerOfTwo(executors.length)){ ret`ur`n newP`o`werOfTwoEventExecutorChooser(executors); }else { retu rn` n`ewGene`r`icEventExecutorChooser(executors); } } 这里设置 的策 略也 很简单：一、若是线程池的线程数量是 2^n，采用下面的方式会高效一些：@Override publicEv entExec utornext() { return ex`e`c`u`tors[idx.getAndIncrement()&executors.length-1]; } 二、若是不是，用取模的方式 ：@ Override publicEventExec utornex t() { returnexecuto rs`\[`M`a`th.abs(idx.getAndIncrement()%executors.length)]; }`

走了这么久，咱们终于到了一个干实事的构造方法中了：

protectedMultithreadEventExecutorGroup(intnThreads,Executorexecutor,
EventExecutorChooserFactorychooserFactory,Object...args){
if(nThreads<=0){
thrownewIllegalArgumentException(String.format("nThreads:%d(expected:>0)",nThreads));
}

// executor 若是是 null，作一次和前面同样的默认设置。
if(executor==null){
executor=newThreadPerTaskExecutor(newDefaultThreadFactory());
}

//这里的children数组很是重要，它就是线程池中的线程数组，这么说不太严谨，可是就大概这个意思
children=newEventExecutor[nThreads];

//下面这个for循环将实例化children数组中的每个元素
for(inti=0;i<nThreads;i++){
booleansuccess=false;
try{
//实例化！！！！！！
children[i]=newChild(executor,args);
success=true;
}catch(Exceptione){
//TODO:Thinkaboutifthisisagoodexceptiontype
thrownewIllegalStateException("failedtocreateachildeventloop",e);
}finally{
//若是有一个child实例化失败，那么success就会为false，而后进入下面的失败处理逻辑
if(!success){
//把已经成功实例化的“线程”shutdown，shutdown是异步操做
for(intj=0;j<i;j++){
children[j].shutdownGracefully();
}

//等待这些线程成功shutdown
for(intj=0;j<i;j++){
EventExecutore=children[j];
try{
while(!e.isTerminated()){
e.awaitTermination(Integer.MAX_VALUE,TimeUnit.SECONDS);
}
}catch(InterruptedExceptioninterrupted){
//把中断状态设置回去，交给关心的线程来处理.
Thread.currentThread().interrupt();
break;
}
}
}
}
}
//================================================
//===到这里，就是表明上面的实例化全部线程已经成功结束===
//================================================

//经过以前设置的chooserFactory来实例化Chooser，把线程池数组传进去，
//这就没必要再说了吧，实现线程选择策略
chooser=chooserFactory.newChooser(children);

//设置一个Listener用来监听该线程池的termination事件
//下面的代码逻辑是：给池中每个线程都设置这个 listener，当监听到全部线程都 terminate 之后，这个线程池就算真正的 terminate 了。
finalFutureListener<Object>terminationListener=newFutureListener<Object>(){
@Override
publicvoidoperationComplete(Future<Object>future)throwsException{
if(terminatedChildren.incrementAndGet()==children.length){
terminationFuture.setSuccess(null);
}
}
};
for(EventExecutore:children){
e.terminationFuture().addListener(terminationListener);
}

//设置readonlyChildren，它是只读集合，之后用到再说
Set<EventExecutor>childrenSet=newLinkedHashSet<EventExecutor>(children.length);
Collections.addAll(childrenSet,children);
readonlyChildren=Collections.unmodifiableSet(childrenSet);
}复制代码

上面的代码很是简单吧，没有什么须要特别说的，接下来，咱们来看看 newChild() 这个方法，这个方法很是重要，它将建立线程池中的线程。

我上面已经用过不少次"线程"这个词了，它可不是 Thread 的意思，而是指池中的个体，后面咱们会看到每一个"线程"在何时会真正建立 Thread 实例。反正每一个 NioEventLoop 实例内部都会有一个本身的 Thread 实例，因此把这两个概念混在一块儿也无所谓吧。

newChild(…) 方法在 NioEventLoopGroup 中覆写了，上面说的"线程"其实就是 NioEventLoop：

@Override
protectedEventLoopnewChild(Executorexecutor,Object...args)throwsException{
returnnewNioEventLoop(this,executor,(SelectorProvider)args[0],
((SelectStrategyFactory)args[1]).newSelectStrategy(),(RejectedExecutionHandler)args[2]);
}

它调用了 NioEventLoop 的构造方法：复制代码

NioEventLoop(NioEventLoopGroupparent,Executorexecutor,SelectorProviderselectorProvider,
SelectStrategystrategy,RejectedExecutionHandlerrejectedExecutionHandler){
//调用父类构造器
super(parent,executor,false,DEFAULT_MAX_PENDING_TASKS,rejectedExecutionHandler);
if(selectorProvider==null){
thrownewNullPointerException("selectorProvider");
}
if(strategy==null){
thrownewNullPointerException("selectStrategy");
}
provider=selectorProvider;
//开启 NIO 中最重要的组件：Selector
finalSelectorTupleselectorTuple=openSelector();
selector=selectorTuple.selector;
unwrappedSelector=selectorTuple.unwrappedSelector;
selectStrategy=strategy;
}复制代码

咱们先粗略观察一下，而后再往下看：

• 在 Netty 中，NioEventLoopGroup 表明线程池，NioEventLoop 就是其中的线程。
• 线程池 NioEventLoopGroup 是池中的线程 NioEventLoop 的 parent，从上面的代码中的取名能够看出。
• 每一个 NioEventLoop 都有本身的 Selector，上面的代码也反应了这一点，这和 Tomcat 中的 NIO 模型有点区别。
• executor、selectStrategy 和 rejectedExecutionHandler 从 NioEventLoopGroup 中一路传到了 NioEventLoop 中。

这个时候，咱们来看一下 NioEventLoop 类的属性都有哪些，咱们先忽略它继承自父类的属性，单单看它本身的：

privateSelectorselector;
privateSelectorunwrappedSelector;
privateSelectedSelectionKeySetselectedKeys;
privatefinalSelectorProviderprovider;
privatefinalAtomicBooleanwakenUp=newAtomicBoolean();
privatefinalSelectStrategyselectStrategy;
privatevolatileintioRatio=50;
privateintcancelledKeys;
privatebooleanneedsToSelectAgain;复制代码

结合它的构造方法咱们来总结一下：

• provider：它由 NioEventLoopGroup 传进来，前面咱们说了一个线程池有一个 selectorProvider，用于建立 Selector 实例
• selector：虽然咱们还没看建立 selector 的代码，但咱们已经知道，在 Netty 中 Selector 是跟着线程池中的线程走的。也就是说，并不是一个线程池一个 Selector 实例，而是线程池中每个线程都有一个 Selector 实例。 在无参构造过程当中，咱们发现，Netty 设置线程个数是 CPU 核心数的两倍，假设咱们的机器 CPU 是 4 核，那么对应的就会有 8 个 Selector 实例。
• selectStrategy：select 操做的策略，这个不急。
• ioRatio：这是 IO 任务的执行时间比例，由于每一个线程既有 IO 任务执行，也有非 IO 任务须要执行，因此该参数为了保证有足够时间是给 IO 的。这里也不须要急着去理解什么 IO 任务、什么非 IO 任务。

而后咱们继续走它的构造方法，咱们看到上面的构造方法调用了父类的构造器，它的父类是 SingleThreadEventLoop。

protectedSingleThreadEventLoop(EventLoopGroupparent,Executorexecutor,
booleanaddTaskWakesUp,intmaxPendingTasks,
RejectedExecutionHandlerrejectedExecutionHandler){
super(parent,executor,addTaskWakesUp,maxPendingTasks,rejectedExecutionHandler);

//咱们能够直接忽略这个东西，之后咱们也不会再介绍它
tailTasks=newTaskQueue(maxPendingTasks);
}

复制代码

SingleThreadEventLoop 这个名字很诡异有没有？而后它的构造方法又调用了父类 SingleThreadEventExecutor 的构造方法：

protectedSingleThreadEventExecutor(EventExecutorGroupparent,Executorexecutor,
booleanaddTaskWakesUp,intmaxPendingTasks,
RejectedExecutionHandlerrejectedHandler){
super(parent);
this.addTaskWakesUp=addTaskWakesUp;
this.maxPendingTasks=Math.max(16,maxPendingTasks);
this.executor=ObjectUtil.checkNotNull(executor,"executor");
//taskQueue，这个东西很重要，提交给NioEventLoop的任务都会进入到这个taskQueue中等待被执行
//这个queue的默认容量是16
taskQueue=newTaskQueue(this.maxPendingTasks);
rejectedExecutionHandler=ObjectUtil.checkNotNull(rejectedHandler,"rejectedHandler");
}复制代码

到这里就更加诡异了，NioEventLoop 的父类是 SingleThreadEventLoop，而 SingleThreadEventLoop 的父类是 SingleThreadEventExecutor，它的名字告诉咱们，它是一个 Executor，是一个线程池，并且是 Single Thread 单线程的。也就是说，线程池 NioEventLoopGroup 中的每个线程 NioEventLoop 也能够当作一个线程池来用，只不过它只有一个线程。这种设计虽然看上去很巧妙，不过有点反人类的样子。上面这个构造函数比较简单：

• 设置了 parent，也就是以前建立的线程池 NioEventLoopGroup 实例
• executor：它是咱们以前实例化的 ThreadPerTaskExecutor，咱们说过，这个东西在线程池中没有用，它是给 NioEventLoop 用的，立刻咱们就要看到它了。提早透露一下，它用来开启 NioEventLoop 中的线程（Thread 实例）。
• taskQueue：这算是该构造方法中新的东西，它是任务队列。咱们前面说过，NioEventLoop 须要负责 IO 事件和非 IO 事件，一般它都在执行 selector 的 select 方法或者正在处理 selectedKeys，若是咱们要 submit 一个任务给它，任务就会被放到 taskQueue 中，等它来轮询。该队列是线程安全的 LinkedBlockingQueue，默认容量为 16。
• rejectedExecutionHandler：taskQueue 的默认容量是 16，因此，若是 submit 的任务堆积了到了 16，再往里面提交任务会触发 rejectedExecutionHandler 的执行策略。

还记得默认策略吗：抛出RejectedExecutionException 异常。 在 NioEventLoopGroup 的默认构造中，它的实现是这样的： private static final`RejectedExecutionHandlerREJECT=newR ejectedExecutionHandler(){ @Ove rride publ ic void rejec ted(Runna bletask,SingleThreadEventExecutorexecutor) { throw` `n`ew`Rejec`t`edExecutionException(); } };`

而后，咱们再回到 NioEventLoop 的构造方法：

NioEventLoop(NioEventLoopGroupparent,Executorexecutor,SelectorProviderselectorProvider,
SelectStrategystrategy,RejectedExecutionHandlerrejectedExecutionHandler){
//咱们刚刚说完了这个
super(parent,executor,false,DEFAULT_MAX_PENDING_TASKS,rejectedExecutionHandler);
if(selectorProvider==null){
thrownewNullPointerException("selectorProvider");
}
if(strategy==null){
thrownewNullPointerException("selectStrategy");
}
provider=selectorProvider;
//建立selector实例
finalSelectorTupleselectorTuple=openSelector();
selector=selectorTuple.selector;
unwrappedSelector=selectorTuple.unwrappedSelector;
selectStrategy=strategy;
}复制代码

能够看到，最重要的方法其实就是 openSelector() 方法，它将建立 NIO 中最重要的一个组件 Selector。在这个方法中，Netty 也作了一些优化，这部分咱们就不去分析它了。到这里，咱们的线程池 NioEventLoopGroup 建立完成了，而且实例化了池中的全部 NioEventLoop 实例。同时，你们应该已经看到，上面并无真正建立 NioEventLoop 中的线程（没有建立 Thread 实例）。提早透露一下，建立线程的时机在第一个任务提交过来的时候，那么第一个任务是什么呢？是咱们立刻要说的 channel 的 register 操做。

365天干货不断微信搜索「猿灯塔」第一时间阅读，回复【资料】【面试】【简历】有我准备的一线大厂面试资料和简历模板