Netty源码—2.Reactor线程模型一

大纲

1.关于NioEventLoop的问题整理

2.理解Reactor线程模型主要分三部分

3.NioEventLoop的创建

4.NioEventLoop的启动

 

1.关于NioEventLoop的问题整理

一.默认下Netty服务端起多少线程及何时启动？

答：默认是2倍CPU核数个线程。在调用EventExcutor的execute(task)方法时，会判断当前线程是否为Netty的Reactor线程，也就是判断当前线程是否为NioEventLoop对应的线程实体。如果是，则说明Netty的Reactor线程已经启动了。如果不是，则说明是外部线程调用EventExcutor的execute()方法。于是会先调用startThread()方法判断当前线程是否已被启动，如果还没有被启动就启动当前线程作为Netty的Reactor线程。

 

二.Netty是如何解决JDK空轮询的？

答：Netty会判断如果当前阻塞的一个Select()操作并没有花那么长时间，那么就说明此时有可能触发了空轮询Bug。默认情况下如果这个现象达到512次，那么就重建一个Selector，并且把之前Selector上所有的key重新移交到新Selector上。通过以上这种处理方式来避免JDK空轮询Bug。

 

三.Netty是如何保证异步串行无锁化的？

答：异步串行无锁化有两个场景。

场景一：拿到客户端一个Channel，不需要对该Channel进行同步，直接就可以多线程并发读写。

场景二：ChannelHandler里的所有操作都是线程安全的，不需要进行同步。

 

Netty在所有外部线程去调用EventLoop或者Channel的方法时，会通过inEventLoop()方法来判断出当前线程是外部线程(非NioEventLoop的线程实体)。在这种情况下，会把所有操作都封装成一个Task放入MPSC队列，然后在NioEventLoop的执行逻辑也就是run()方法里，这些Task会被逐个执行。

 

2.理解Reactor线程模型主要分三部分

一.NioEventLoop的创建

二.NioEventLoop的启动

三.NioEventLoop的执行

 

3.NioEventLoop的创建

(1)创建入口

(2)确定NioEventLoop的个数

(3)NioEventLoopGroup的创建流程

(4)创建线程执行器ThreadPerTaskExecutor

(5)创建NioEventLoop

(6)创建线程选择器EventExecutorChooser

(7)NioEventLoopGroup的创建总结

 

(1)创建入口

EventLoopGroup bossGroup = new NioEventLoopGroup(); EventLoopGroup workerGroup = new NioEventLoopGroup();

(2)确定NioEventLoop的个数

由NioEventLoopGroup的构造方法来确定NioEventLoop的个数。如果NioEventLoopGroup没有传递构造参数，那么NioEventLoop线程的个数为CPU核数的2倍。如果NioEventLoopGroup传递了参数n，那么NioEventLoop线程的个数就是n。

 

(3)NioEventLoopGroup的创建流程

NioEventLoopGroup的构造方法会触发创建流程。

一.创建线程执行器ThreadPerTaskExecutor

每次调用ThreadPerTaskExecutor.execute()方法时都会创建一个线程。

二.创建NioEventLoop

NioEventLoop对应NioEventLoopGroup线程池里的线程，NioEventLoopGroup的构造方法会用一个for循环通过调用newChild()方法来创建NioEventLoop线程。

三.创建线程选择器EventExecutorChooser

线程选择器的作用是用于给每个新连接分配一个NioEventLoop线程，也就是从NioEventLoopGroup线程池中选择一个NioEventLoop线程来处理新连接。

//MultithreadEventLoopGroup implementations which is used for NIO Selector based Channels. public class NioEventLoopGroup extends MultithreadEventLoopGroup {      //Create a new instance using the default number of threads,      //the default ThreadFactory and the SelectorProvider which is returned by SelectorProvider#provider().     public NioEventLoopGroup() {         this(0);     }          //Create a new instance using the specified number of threads,      //ThreadFactory and the SelectorProvider which is returned by SelectorProvider#provider().     public NioEventLoopGroup(int nThreads) {         this(nThreads, (Executor) null);     }          public NioEventLoopGroup(int nThreads, Executor executor) {         this(nThreads, executor, SelectorProvider.provider());     }          public NioEventLoopGroup(int nThreads, Executor executor, final SelectorProvider selectorProvider) {         this(nThreads, executor, selectorProvider, DefaultSelectStrategyFactory.INSTANCE);     }          public NioEventLoopGroup(int nThreads, Executor executor, final SelectorProvider selectorProvider, final SelectStrategyFactory selectStrategyFactory) {          super(nThreads, executor, selectorProvider, selectStrategyFactory, RejectedExecutionHandlers.reject());     }     ... }  //Abstract base class for EventLoopGroup implementations that handles their tasks with multiple threads at the same time. public abstract class MultithreadEventLoopGroup extends MultithreadEventExecutorGroup implements EventLoopGroup {     private static final InternalLogger logger = InternalLoggerFactory.getInstance(MultithreadEventLoopGroup.class);     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);     }      protected MultithreadEventLoopGroup(int nThreads, Executor executor, Object... args) {         super(nThreads == 0 ? DEFAULT_EVENT_LOOP_THREADS : nThreads, executor, args);     }     ... }  //Abstract base class for EventExecutorGroup implementations that handles their tasks with multiple threads at the same time. public abstract class MultithreadEventExecutorGroup extends AbstractEventExecutorGroup {     private final EventExecutor[] children;     private final EventExecutorChooserFactory.EventExecutorChooser chooser;     ...     //Create a new instance.     protected MultithreadEventExecutorGroup(int nThreads, Executor executor, Object... args) {         this(nThreads, executor, DefaultEventExecutorChooserFactory.INSTANCE, args);     }          //Create a new instance.     //@param nThreads，the number of threads that will be used by this instance.     //@param executor，the Executor to use, or null if the default should be used.     //@param chooserFactory，the EventExecutorChooserFactory to use.     //@param args，arguments which will passed to each #newChild(Executor, Object...) call     protected MultithreadEventExecutorGroup(int nThreads, Executor executor, EventExecutorChooserFactory chooserFactory, Object... args) {         if (nThreads <= 0) throw new IllegalArgumentException(String.format("nThreads: %d (expected: > 0)", nThreads));         //1.创建ThreadPerTaskExecutor线程执行器         if (executor == null) executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());         //2.创建NioEventLoop         children = new EventExecutor[nThreads];         for (int i = 0; i < nThreads; i ++) {             ...             //创建每一个NioEventLoop时，都会调用newChild()方法给每一个NioEventLoop配置一些核心参数             //传入线程执行器executor去创建NioEventLoop             children[i] = newChild(executor, args);         }         //3.创建线程选择器         chooser = chooserFactory.newChooser(children);         ...     }     ... }

创建NioEventLoopGroup的脉络如下：

new NioEventLoopGroup() //线程组，线程个数默认为2 * CPU核数   new ThreadPerTaskExecutor() //创建线程执行器，作用是负责创建NioEventLoop对应的线程   for(...) { newChild() } //构造NioEventLoop，创建NioEventLoop线程组   chooserFactory.newChooser() //线程选择器，用于给每个新连接分配一个NioEventLoop线程

(4)创建线程执行器ThreadPerTaskExecutor

ThreadPerTaskExecutor的作用是：每次调用它的execute()方法执行Runnable任务时，都会通过threadFactory.newThread()创建出一个线程，然后把要执行的Runnable任务传递进该线程进行执行。

 

其中成员变量threadFactory是在传参给ThreadPerTaskExecutor的构造方法时，由newDefaultThreadFactory()方法构建的，也就是一个DefaultThreadFactory对象。

 

所以线程执行器ThreadPerTaskExecutor在通过threadFactory.newThread()创建线程时，其实就是调用DefaultThreadFactory的newThread()方法。

 

而DefaultThreadFactory.newThread()方法创建出来的线程实体，是Netty经过优化之后的FastThreadLocalThread对象，这个线程实体在操作ThreadLocal时，要比JDK快。

 

ThreadPerTaskExecutor线程执行器总结：

一.每次执行ThreadPerTaskExecutor的execute()方法时，都会创建出一个FastThreadLocalThread的线程实体，所以Netty的线程实体都是由ThreadPerTaskExecutor创建的。

二.FastThreadLocalThread线程实体的命名规则是：nioEventLoop-自增的线程池编号-自增的线程数编号。

//Abstract base class for EventExecutorGroup implementations that handles their tasks with multiple threads at the same time. public abstract class MultithreadEventExecutorGroup extends AbstractEventExecutorGroup {     private final EventExecutor[] children;     private final EventExecutorChooserFactory.EventExecutorChooser chooser;     ...         //Create a new instance.     protected MultithreadEventExecutorGroup(int nThreads, Executor executor, EventExecutorChooserFactory chooserFactory, Object... args) {         if (nThreads <= 0) throw new IllegalArgumentException(String.format("nThreads: %d (expected: > 0)", nThreads));         //1.创建ThreadPerTaskExecutor线程执行器         if (executor == null) executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());         //2.创建NioEventLoop         children = new EventExecutor[nThreads];         for (int i = 0; i < nThreads; i ++) {             ...             //创建每一个NioEventLoop时，都会调用newChild()方法给每一个NioEventLoop配置一些核心参数             //传入线程执行器executor去创建NioEventLoop             children[i] = newChild(executor, args);         }         //3.创建线程选择器         chooser = chooserFactory.newChooser(children);         ...     }          protected ThreadFactory newDefaultThreadFactory() {         //getClass()是获取该方法所属的对象类型，也就是NioEventLoopGroup类型         //因为是通过NioEventLoopGroup的构造方法层层调用到这里的         return new DefaultThreadFactory(getClass());     }     ... }  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) {         //调用DefaultThreadFactory的newThread()方法执行Runnable任务         threadFactory.newThread(command).start();     } }  //A ThreadFactory implementation with a simple naming rule. public class DefaultThreadFactory implements ThreadFactory {     private static final AtomicInteger poolId = new AtomicInteger();     private final AtomicInteger nextId = new AtomicInteger();     private final boolean daemon;     private final int priority;     protected final ThreadGroup threadGroup;     ...     public DefaultThreadFactory(Class<?> poolType) {         this(poolType, false, Thread.NORM_PRIORITY);     }          public DefaultThreadFactory(Class<?> poolType, boolean daemon, int priority) {         //toPoolName()方法会把NioEventLoopGroup的首字母变成小写         this(toPoolName(poolType), daemon, priority);     }          public DefaultThreadFactory(String poolName, boolean daemon, int priority) {         this(poolName, daemon, priority,          System.getSecurityManager() == null? Thread.currentThread().getThreadGroup() : System.getSecurityManager().getThreadGroup());     }          public DefaultThreadFactory(String poolName, boolean daemon, int priority, ThreadGroup threadGroup) {         ...         //prefix用来标记线程名字的前缀         prefix = poolName + '-' + poolId.incrementAndGet() + '-';         this.daemon = daemon;         this.priority = priority;         this.threadGroup = threadGroup;     }          @Override     public Thread newThread(Runnable r) {         Thread t = newThread(new DefaultRunnableDecorator(r), prefix + nextId.incrementAndGet());         if (t.isDaemon()) {             if (!daemon) t.setDaemon(false);         } else {             if (daemon) t.setDaemon(true);         }         if (t.getPriority() != priority) t.setPriority(priority);         return t;     }          protected Thread newThread(Runnable r, String name) {         return new FastThreadLocalThread(threadGroup, r, name);     }     ... }

(5)创建NioEventLoop

说明一：

由MultithreadEventExecutorGroup的构造方法可知，Netty会使用for循环 + newChild()方法来创建nThreads个NioEventLoop，而且一个NioEventLoop对应一个线程实体FastThreadLocalThread。

//Abstract base class for EventExecutorGroup implementations that handles their tasks with multiple threads at the same time. public abstract class MultithreadEventExecutorGroup extends AbstractEventExecutorGroup {     private final EventExecutor[] children;     private final EventExecutorChooserFactory.EventExecutorChooser chooser;     ...         //Create a new instance.     protected MultithreadEventExecutorGroup(int nThreads, Executor executor, EventExecutorChooserFactory chooserFactory, Object... args) {         if (nThreads <= 0) throw new IllegalArgumentException(String.format("nThreads: %d (expected: > 0)", nThreads));         //1.创建ThreadPerTaskExecutor线程执行器         if (executor == null) executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());         //2.创建NioEventLoop         children = new EventExecutor[nThreads];         for (int i = 0; i < nThreads; i ++) {             ...             //创建每一个NioEventLoop时，都会调用newChild()方法给每一个NioEventLoop配置一些核心参数             //传入线程执行器executor去创建NioEventLoop             children[i] = newChild(executor, args);         }         //3.创建线程选择器         chooser = chooserFactory.newChooser(children);         ...     }          //Create a new EventExecutor which will later then accessible via the #next() method.     //This method will be called for each thread that will serve this MultithreadEventExecutorGroup.     protected abstract EventExecutor newChild(Executor executor, Object... args) throws Exception;     ... }

说明二：

MultithreadEventExecutorGroup的newChild()抽象方法是由NioEventLoopGroup实现的，所以在执行NioEventLoopGroup的默认构造方法时也会执行其newChild()方法。

 

NioEventLoopGroup的newChild()方法需要传递一个executor参数，该参数就是执行NioEventLoopGroup构造方法开始时创建的线程执行器，之后newChild()方法会返回一个新创建的NioEventLoop对象。

//MultithreadEventLoopGroup implementations which is used for NIO Selector based Channels. public class NioEventLoopGroup extends MultithreadEventLoopGroup {     ...     @Override     protected EventLoop newChild(Executor executor, Object... args) throws Exception {         //executor是执行NioEventLoopGroup构造方法开始时创建的线程执行器ThreadPerTaskExecutor         //this指的是NioEventLoopGroup，表示新创建的NioEventLoop对象归属哪个NioEventLoopGroup         return new NioEventLoop(this, executor, (SelectorProvider) args[0],             ((SelectStrategyFactory) args[1]).newSelectStrategy(), (RejectedExecutionHandler) args[2]);     }     ... }

说明三：

创建NioEventLoop对象时，NioEventLoop的构造方法会通过调用其openSelector()方法来创建一个Selector，所以一个Selector就和一个NioEventLoop绑定了，而一个Selector可以将多个连接绑定在一起来负责监听这些连接的读写事件。

 

在NioEventLoop的openSelector()方法中，Netty会通过反射对Selector底层的数据结构进行优化(Hash Set => 数组)。

//SingleThreadEventLoop implementation which register the Channel's to a Selector and so does the multi-plexing of these in the event loop. public final class NioEventLoop extends SingleThreadEventLoop {     //The NIO Selector.     Selector selector;     private final SelectorProvider provider;     private final SelectStrategy selectStrategy;     ...     NioEventLoop(NioEventLoopGroup parent, Executor executor, SelectorProvider selectorProvider,         SelectStrategy strategy, RejectedExecutionHandler rejectedExecutionHandler) {         //调用其父类SingleThreadEventLoop的构造方法         super(parent, executor, false, DEFAULT_MAX_PENDING_TASKS, rejectedExecutionHandler);         if (selectorProvider == null) throw new NullPointerException("selectorProvider");         if (strategy == null) throw new NullPointerException("selectStrategy");         this.provider = selectorProvider;         this.selector = openSelector();//创建一个Selector         this.selectStrategy = strategy;     }          private Selector openSelector() {         final Selector selector;         try {             selector = provider.openSelector();             ...         } catch(IOException e) {             ...         }         ...         return selector;     }     ... }

说明四：

NioEventLoop的构造方法还会调用其父类的父类SingleThreadEventExecutor的构造方法。SingleThreadEventExecutor的构造方法里有两个关键的操作：一是把线程执行器保存起来，因为后面创建NioEventLoop对应的线程时要用到。二是创建一个MPSC任务队列，因为Netty中所有异步执行的本质都是通过该任务队列来协调完成的。

//Abstract base class for EventLoops that execute all its submitted tasks in a single thread. public abstract class SingleThreadEventLoop extends SingleThreadEventExecutor implements EventLoop {     private final Queue<Runnable> tailTasks;     ...     protected SingleThreadEventLoop(EventLoopGroup parent, Executor executor,          boolean addTaskWakesUp, int maxPendingTasks, RejectedExecutionHandler rejectedExecutionHandler) {         //调用其父类SingleThreadEventExecutor的构造方法         super(parent, executor, addTaskWakesUp, maxPendingTasks, rejectedExecutionHandler);         //调用父类SingleThreadEventExecutor的newTaskQueue()方法         tailTasks = newTaskQueue(maxPendingTasks);     }     ... }  //Abstract base class for OrderedEventExecutor's that execute all its submitted tasks in a single thread. public abstract class SingleThreadEventExecutor extends AbstractScheduledEventExecutor implements OrderedEventExecutor {     private final boolean addTaskWakesUp;     private final Executor executor;     private final int maxPendingTasks;     private final Queue<Runnable> taskQueue;     private final RejectedExecutionHandler rejectedExecutionHandler;     ...     //Create a new instance     protectedSingleThreadEventExecutor(EventExecutorGroup parent, ThreadFactory threadFactory,           boolean addTaskWakesUp, int maxPendingTasks, RejectedExecutionHandler rejectedHandler) {         this(parent, new ThreadPerTaskExecutor(threadFactory), addTaskWakesUp, maxPendingTasks, rejectedHandler);     }          //Create a new instance     //@param parent，the EventExecutorGroup which is the parent of this instance and belongs to it     //@param executor，the Executor which will be used for executing     //@param addTaskWakesUp，true if and only if invocation of #addTask(Runnable) will wake up the executor thread     //@param maxPendingTasks，the maximum number of pending tasks before new tasks will be rejected.     //@param rejectedHandler，the RejectedExecutionHandler to use.     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");         //关键操作二：创建一个MPSC任务队列         this.taskQueue = newTaskQueue(this.maxPendingTasks);         this.rejectedExecutionHandler = ObjectUtil.checkNotNull(rejectedHandler, "rejectedHandler");     }          //Create a new Queue which will holds the tasks to execute.     //NioEventLoop会重写这个newTaskQueue()方法     protected Queue<Runnable> newTaskQueue(int maxPendingTasks) {         return new LinkedBlockingQueue<Runnable>(maxPendingTasks);     }     ... }  //SingleThreadEventLoop implementation which register the Channel's to a Selector and so does the multi-plexing of these in the event loop. public final class NioEventLoop extends SingleThreadEventLoop {     ...     //创建一个MPSC任务队列     @Override     protected Queue<Runnable> newTaskQueue(int maxPendingTasks) {         //This event loop never calls takeTask()         return PlatformDependent.newMpscQueue(maxPendingTasks);     }     ... }

MPSC队列也就是多生产者单消费者队列。单消费者是指某个NioEventLoop对应的线程(执行其run()方法的那个线程)。多生产者就是这个NioEventLoop对应的线程之外的线程，通常情况下就是我们的业务线程。比如，一些线程在调用writeAndFlush()方法时可以不用考虑线程安全而随意调用，那么这些线程就是多生产者。

 

MPSC队列是通过JCTools这个工具包来实现的，Netty的高性能很大程度上要归功于这个工具包。MPSC的全称是Muti Producer Single Consumer。Muti Producer对应的是外部线程，Single Consumer对应的是Netty的NioEventLoop线程。外部线程在执行Netty的一些任务时，如果判断不是由NioEventLoop对应的线程执行的，就会直接放入一个任务队列里，然后由一个NioEventLoop对应的线程去执行。

 

创建NioEventLoop总结：

NioEventLoopGroup的newChild()方法创建NioEventLoop时做了三项事情：一.创建一个Selector用于轮询注册到该NioEventLoop上的连接，二.创建一个MPSC任务队列，三.保存线程执行器到NioEventLoop。

 

(6)创建线程选择器EventExecutorChooser

说明一：

在传统的BIO编程中，一个新连接被创建后，通常需要给这个连接绑定一个Selector，之后这个连接的整个生命周期都由这个Selector管理。

 

说明二：

创建NioEventLoop时会创建一个Selector，所以一个Selector会对应一个NioEventLoop，一个NioEventLoop上会有一个Selector。线程选择器的作用就是为一个连接在NioEventLoopGroup中选择一个NioEventLoop，从而将该连接绑定到这个NioEventLoop的Selector上。

 

说明三：

根据MultithreadEventExecutorGroup的构造方法，会使用DefaultEventExecutorChooserFactory的newChooser()方法来创建线程选择器。创建好线程选择器EventExecutorChooser之后，便可以通过其next()方法获取一个NioEventLoop。

 

Netty通过判断NioEventLoopGroup中的NioEventLoop个数是否是2的幂来创建不同的线程选择器。但不管是哪一种选择器，最终效果都是从第一个NioEventLoop开始遍历到最后一个NioEventLoop，然后再从第一个NioEventLoop开始，如此循环。

//Abstract base class for EventExecutorGroup implementations that handles their tasks with multiple threads at the same time. public abstract class MultithreadEventExecutorGroup extends AbstractEventExecutorGroup {     private final EventExecutor[] children;     private final EventExecutorChooserFactory.EventExecutorChooser chooser;     ...     //Create a new instance.     protected MultithreadEventExecutorGroup(int nThreads, Executor executor, Object... args) {         this(nThreads, executor, DefaultEventExecutorChooserFactory.INSTANCE, args);     }          //Create a new instance.     //@param nThreads，the number of threads that will be used by this instance.     //@param executor，the Executor to use, or null if the default should be used.     //@param chooserFactory，the EventExecutorChooserFactory to use.     //@param args，arguments which will passed to each #newChild(Executor, Object...) call     protected MultithreadEventExecutorGroup(int nThreads, Executor executor, EventExecutorChooserFactory chooserFactory, Object... args) {         if (nThreads <= 0) throw new IllegalArgumentException(String.format("nThreads: %d (expected: > 0)", nThreads));         //1.创建ThreadPerTaskExecutor线程执行器         if (executor == null) executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());         //2.创建NioEventLoop         children = new EventExecutor[nThreads];         for (int i = 0; i < nThreads; i ++) {             ...             //创建每一个NioEventLoop时，都会调用newChild()方法给每一个NioEventLoop配置一些核心参数             //传入线程执行器executor去创建NioEventLoop             children[i] = newChild(executor, args);         }         //3.创建线程选择器，chooserFactory就是传入的DefaultEventExecutorChooserFactory实例         chooser = chooserFactory.newChooser(children);         ...     }     ... }  //Default implementation which uses simple round-robin to choose next EventExecutor. @UnstableApi 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)) {             //如果NioEventLoop个数是2的幂，则进行位与运算             return new PowerOfTowEventExecutorChooser(executors);         } else {             //如果NioEventLoop个数不是2的幂，则进行普通取模运算             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)];         }     } }

说明四：

创建NioEventLoopGroup的最后一个步骤就是创建线程选择器chooser，创建线程选择器的流程如下：

chooserFactory.newChooser() //创建线程选择器的入口，chooser的作用就是为新连接绑定一个NioEventLoop   DefaultEventExecutorChooserFactory.isPowerOfTwo() //判断NioEventLoop个数是否为2的幂     PowerOfTowEventExecutorChooser //优化       index++ & (length - 1) //位与运算     GenericEventExecutorChooser //普通       abs(index++ % length) //取模运算

(7)NioEventLoopGroup的创建总结

默认情况下，NioEventLoopGroup会创建2倍CPU核数个NioEventLoop。一个NioEventLoop和一个Selector以及一个MPSC任务队列一一对应。

 

NioEventLoop线程的命名规则是nioEventLoopGroup-xx-yy，其中xx表示全局第xx个NioEventLoopGroup线程池，yy表示这个NioEventLoop在这个NioEventLoopGroup中是属于第yy个。

 

线程选择器chooser的作用是为一个连接选择一个NioEventLoop，可通过线程选择器的next()方法返回一个NioEventLoop。如果NioEventLoop的个数为2的幂，则next()方法会使用位与运算进行优化。

 

一个NioEventLoopGroup会有一个线程执行器executor、一个线程选择器chooser、一个数组children存放2倍CPU核数个NioEventLoop。

 

4.NioEventLoop的启动

(1)启动NioEventLoop的两大入口

(2)判断当前线程是否是NioEventLoop线程

(3)创建一个线程并启动

(4)NioEventLoop的启动总结

 

(1)启动NioEventLoop的两大入口

入口一：服务端启动，注册服务端Channel到Selector时

入口二：新连接接入，通过chooser绑定一个NioEventLoop时

 

下面先看入口一：

 

调用ServerBootstrap的bind()方法其实会调用AbstractBootstrap的doBind()方法，然后会调用AbstractBootstrap的initAndRegister()方法，接着执行config().group().register(channel)注册服务端Channel。最后会逐层深入调用到AbstractChannel.AbstractUnsafe的register()方法，来启动一个NioEventLoop将服务端Channel注册到Selector上。

//Bootstrap sub-class which allows easy bootstrap of ServerChannel public class ServerBootstrap extends AbstractBootstrap<ServerBootstrap, ServerChannel> {     ...     ... }  //AbstractBootstrap is a helper class that makes it easy to bootstrap a Channel.  //It support method-chaining to provide an easy way to configure the AbstractBootstrap. //When not used in a ServerBootstrap context, the #bind() methods are useful for connectionless transports such as datagram (UDP). public abstract class AbstractBootstrap<B extends AbstractBootstrap<B, C>, C extends Channel> implements Cloneable {     ...     //Create a new Channel and bind it.     public ChannelFuture bind(int inetPort) {         //首先根据端口号创建一个InetSocketAddress对象，然后调用重载方法bind()         return bind(new InetSocketAddress(inetPort));     }          //Create a new Channel and bind it.     public ChannelFuture bind(SocketAddress localAddress) {         //验证服务启动需要的必要参数         validate();         if (localAddress == null) throw new NullPointerException("localAddress");         return doBind(ObjectUtil.checkNotNull(localAddress, "localAddress"));     }          private ChannelFuture doBind(final SocketAddress localAddress) {         final ChannelFuture regFuture = initAndRegister();//1.初始化和注册Channel         final Channel channel = regFuture.channel();         ...         doBind0(regFuture, channel, localAddress, promise);//2.绑定服务端端口         ...         return promise;     }          final ChannelFuture initAndRegister() {         Channel channel = null;         ...         //1.创建服务端Channel         channel = channelFactory.newChannel();         //2.初始化服务端Channel         init(channel);         ...         //3.注册服务端Channel，比如通过NioEventLoopGroup的register()方法进行注册         ChannelFuture regFuture = config().group().register(channel);         ...         return regFuture;     }     ... }  //Bootstrap sub-class which allows easy bootstrap of ServerChannel public class ServerBootstrap extends AbstractBootstrap<ServerBootstrap, ServerChannel> {     private final ServerBootstrapConfig config = new ServerBootstrapConfig(this);     ...     @Override     public final ServerBootstrapConfig config() {         return config;     }     ... }  public abstract class AbstractBootstrapConfig<B extends AbstractBootstrap<B, C>, C extends Channel> {     protected final B bootstrap;     ...     protected AbstractBootstrapConfig(B bootstrap) {         this.bootstrap = ObjectUtil.checkNotNull(bootstrap, "bootstrap");     }          //Returns the configured EventLoopGroup or null if non is configured yet.     public final EventLoopGroup group() {         //比如返回一个NioEventLoopGroup对象         return bootstrap.group();     }     ... }  //MultithreadEventLoopGroup implementations which is used for NIO Selector based Channels. public class NioEventLoopGroup extends MultithreadEventLoopGroup {     ...     ... }  //Abstract base class for EventLoopGroup implementations that handles their tasks with multiple threads at the same time. public abstract class MultithreadEventLoopGroup extends MultithreadEventExecutorGroup implements EventLoopGroup {     ...     @Override     public ChannelFuture register(Channel channel) {         //先通过next()方法获取一个NioEventLoop，然后通过NioEventLoop.register()方法注册服务端Channel         return next().register(channel);     }          @Override     public EventLoop next() {         return (EventLoop) super.next();     }     ... }  //Abstract base class for EventExecutorGroup implementations that handles their tasks with multiple threads at the same time. public abstract class MultithreadEventExecutorGroup extends AbstractEventExecutorGroup {     private final EventExecutorChooserFactory.EventExecutorChooser chooser;     ...     @Override     public EventExecutor next() {         //通过线程选择器chooser选择一个NioEventLoop         return chooser.next();     }         ... }  //SingleThreadEventLoop implementation which register the Channel's to a Selector and so does the multi-plexing of these in the event loop. public final class NioEventLoop extends SingleThreadEventLoop {     ...     ... }  //Abstract base class for EventLoops that execute all its submitted tasks in a single thread. public abstract class SingleThreadEventLoop extends SingleThreadEventExecutor implements EventLoop {     ...     @Override     public ChannelFuture register(Channel channel) {         return register(new DefaultChannelPromise(channel, this));     }      @Override     public ChannelFuture register(final ChannelPromise promise) {         ObjectUtil.checkNotNull(promise, "promise");         //调用AbstractUnsafe的register()方法         promise.channel().unsafe().register(this, promise);         return promise;     }     ... }  //A skeletal Channel implementation. public abstract class AbstractChannel extends DefaultAttributeMap implements Channel {     private volatile EventLoop eventLoop;     ...     //Unsafe implementation which sub-classes must extend and use.     protected abstract class AbstractUnsafe implements Unsafe {         ...         @Override         public final void register(EventLoop eventLoop, final ChannelPromise promise) {             ...             //绑定事件循环器，即绑定一个NioEventLoop到该Channel上             AbstractChannel.this.eventLoop = eventLoop;             //注册Selector，并启动一个NioEventLoop             if (eventLoop.inEventLoop()) {                 register0(promise);             } else {                 ...                 //通过启动这个NioEventLoop线程来调用register0()方法将这个服务端Channel注册到Selector上                 //其实执行的是SingleThreadEventExecutor的execute()方法                 eventLoop.execute(new Runnable() {                     @Override                     public void run() {                         register0(promise);                     }                 });                 ...             }         }     }     ... }

(2)判断当前线程是否是NioEventLoop线程

调用NioEventLoop的inEventLoop()方法可以判断当前线程是否是Netty的Reactor线程，也就是NioEventLoop对应的线程实体。NioEventLoop的线程实体被创建之后，会将该线程实体保存到NioEventLoop父类的成员变量thread中。

 

服务端启动、注册服务端Channel到Selector，执行到AbstractUnsafe.register()方法中的eventLoop.inEventLoop()代码时，会将main方法对应的主线程传递进来与this.thread进行比较。由于this.thread此时并未赋值，所以为空，因此inEventLoop()方法返回false，于是便会执行eventLoop.execute()代码创建一个线程并启动。

//SingleThreadEventLoop implementation which register the Channel's  //to a Selector and so does the multi-plexing of these in the event loop. public final class NioEventLoop extends SingleThreadEventLoop {     ...     ... }  //Abstract base class for EventLoops that execute all its submitted tasks in a single thread. public abstract class SingleThreadEventLoop extends SingleThreadEventExecutor implements EventLoop {     ...     ... }  //Abstract base class for OrderedEventExecutor's that execute all its submitted tasks in a single thread. public abstract class SingleThreadEventExecutor extends AbstractScheduledEventExecutor implements OrderedEventExecutor {     ...     ... }  //Abstract base class for EventExecutors that want to support scheduling. public abstract class AbstractScheduledEventExecutor extends AbstractEventExecutor {     ...     ... }  //Abstract base class for {@link EventExecutor} implementations. public abstract class AbstractEventExecutor extends AbstractExecutorService implements EventExecutor {     ...     @Override     public boolean inEventLoop() {         //注册服务端Channel时是通过主线程进行注册的，Thread.currentThread()对应的就是main线程         //调用SingleThreadEventExecutor.inEventLoop()方法         return inEventLoop(Thread.currentThread());     }     ... }  //Abstract base class for OrderedEventExecutor's that execute all its submitted tasks in a single thread. public abstract class SingleThreadEventExecutor extends AbstractScheduledEventExecutor implements OrderedEventExecutor {     private volatile Thread thread;     ...     @Override     public boolean inEventLoop(Thread thread) {         return thread == this.thread;//此时线程还没创建，this.thread为null     }     ... }

(3)创建一个线程并启动

AbstractUnsafe.register()方法准备将服务端Channel注册到Selector上时，首先在判断条件中执行eventLoop.inEventLoop()代码发现为false，于是便执行eventLoop.execute()代码创建一个线程并启动它去执行注册任务。而执行eventLoop.execute()代码其实就是调用SingleThreadEventExecutor的execute()方法。

//Abstract base class for EventLoops that execute all its submitted tasks in a single thread. public abstract class SingleThreadEventLoop extends SingleThreadEventExecutor implements EventLoop {     ...     @Override     public ChannelFuture register(Channel channel) {         return register(new DefaultChannelPromise(channel, this));     }      @Override     public ChannelFuture register(final ChannelPromise promise) {         ObjectUtil.checkNotNull(promise, "promise");         //调用AbstractUnsafe的register()方法，并把NioEventLoop自己当作参数传入         promise.channel().unsafe().register(this, promise);         return promise;     }     ... }  //A skeletal Channel implementation. public abstract class AbstractChannel extends DefaultAttributeMap implements Channel {     private volatile EventLoop eventLoop;     ...     //Unsafe implementation which sub-classes must extend and use.     protected abstract class AbstractUnsafe implements Unsafe {         ...         @Override         public final void register(EventLoop eventLoop, final ChannelPromise promise) {             ...             //绑定事件循环器，即绑定一个NioEventLoop到该Channel上             AbstractChannel.this.eventLoop = eventLoop;             //注册Selector，并启动一个NioEventLoop             if (eventLoop.inEventLoop()) {                 register0(promise);             } else {                 ...                 //通过启动这个NioEventLoop线程来调用register0()方法将这个服务端Channel注册到Selector上                 //其实执行的是SingleThreadEventExecutor的execute()方法                 eventLoop.execute(new Runnable() {                     @Override                     public void run() {                         register0(promise);                     }                 });                 ...             }         }     }     ... }  //Abstract base class for OrderedEventExecutor's that execute all its submitted tasks in a single thread. public abstract class SingleThreadEventExecutor extends AbstractScheduledEventExecutor implements OrderedEventExecutor {     private volatile Thread thread;     //创建NioEventLoop时会通过构造方法传入NioEventLoopGroup的线程执行器executor     private final Executor executor;     ...     @Override     public void execute(Runnable task) {         if (task == null) throw new NullPointerException("task");         boolean inEventLoop = inEventLoop();         //判断当前线程是否是Netty的Reactor线程         if (inEventLoop) {             addTask(task);         } else {             startThread();             addTask(task);             if (isShutdown() && removeTask(task)) reject();         }         if (!addTaskWakesUp && wakesUpForTask(task)) wakeup(inEventLoop);     }          private void startThread() {         //判断Reactor线程有没有被启动；如果没有被启动，则通过CAS调用doStartThread()方法启动线程         if (STATE_UPDATER.get(this) == ST_NOT_STARTED) {             if (STATE_UPDATER.compareAndSet(this, ST_NOT_STARTED, ST_STARTED)) {                 doStartThread();             }         }     }      private void doStartThread() {         assert thread == null;         //executor.execute()方法会创建出一个FastThreadLocalThread线程来执行Runnable任务         //所以在Runnable的run()方法中，Thread.currentThread()指的是这个FastThreadLocalThread线程         executor.execute(new Runnable() {             @Override             public void run() {                 //Thread.currentThread()指的是FastThreadLocalThread线程                 thread = Thread.currentThread();                 ...                 SingleThreadEventExecutor.this.run();//启动线程                 ...             }         });     }          //具体的run()方法由子类比如NioEventLoop来实现     protected abstract void run();     ... }

SingleThreadEventExecutor的execute()方法的说明如下：

 

一.这个方法也可能会被用户代码使用，如ctx.executor().execute(task)。所以execute()方法里又调用inEventLoop()方法进行了一次外部线程判断，确保执行task任务时不会遇到线程问题。

 

二.如果当前线程不是Netty的Reactor线程，则调用startThread()方法启动一个Reactor线程。在startThread()方法中首先会判断当前NioEventLoop对应的Reactor线程实体有没有被启动。如果没有被启动，则通过设置CAS成功后调用doStartThread()方法启动线程。

 

三.执行doStartThread()方法时，会调用NioEventLoop的内部成员变量executor的execute()方法。executor就是线程执行器ThreadPerTaskExecutor，它的作用是每次执行Runnable任务时都会创建一个线程来执行。也就是executor.execute()方法会通过DefaultThreadFactory的newThread()方法，创建出一个FastThreadLocalThread线程来执行Runnable任务。

 

四.doStartThread()方法的Runnable任务会由一个FastThreadLocalThread线程来执行。在Runnable任务的run()方法里，会保存ThreadPerTaskExecutor创建出来的FastThreadLocalThread对象到SingleThreadEventExecutor的成员变量thread中，然后调用SingleThreadEventExecutor的run()方法。

//Abstract base class for EventExecutorGroup implementations that handles their tasks with multiple threads at the same time. public abstract class MultithreadEventExecutorGroup extends AbstractEventExecutorGroup {     private final EventExecutor[] children;     private final EventExecutorChooserFactory.EventExecutorChooser chooser;     ...         //Create a new instance.     protected MultithreadEventExecutorGroup(int nThreads, Executor executor, EventExecutorChooserFactory chooserFactory, Object... args) {         if (nThreads <= 0) throw new IllegalArgumentException(String.format("nThreads: %d (expected: > 0)", nThreads));         //1.创建ThreadPerTaskExecutor线程执行器         if (executor == null) executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());         //2.创建NioEventLoop         children = new EventExecutor[nThreads];         for (int i = 0; i < nThreads; i ++) {             ...             //创建每一个NioEventLoop时，都会调用newChild()方法给每一个NioEventLoop配置一些核心参数             //传入线程执行器executor去创建NioEventLoop             children[i] = newChild(executor, args);         }         //3.创建线程选择器         chooser = chooserFactory.newChooser(children);         ...     }          protected ThreadFactory newDefaultThreadFactory() {         //getClass()是获取该方法所属的对象类型，也就是NioEventLoopGroup类型         //因为是通过NioEventLoopGroup的构造方法层层调用到这里的         return new DefaultThreadFactory(getClass());     }     ... }  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) {         //调用DefaultThreadFactory的newThread()方法执行Runnable任务         threadFactory.newThread(command).start();     } }  //A ThreadFactory implementation with a simple naming rule. public class DefaultThreadFactory implements ThreadFactory {     private static final AtomicInteger poolId = new AtomicInteger();     private final AtomicInteger nextId = new AtomicInteger();     private final boolean daemon;     private final int priority;     protected final ThreadGroup threadGroup;     ...     public DefaultThreadFactory(Class<?> poolType) {         this(poolType, false, Thread.NORM_PRIORITY);     }          public DefaultThreadFactory(Class<?> poolType, boolean daemon, int priority) {         //toPoolName()方法会把NioEventLoopGroup的首字母变成小写         this(toPoolName(poolType), daemon, priority);     }          public DefaultThreadFactory(String poolName, boolean daemon, int priority) {         this(poolName, daemon, priority,             System.getSecurityManager() == null ? Thread.currentThread().getThreadGroup() : System.getSecurityManager().getThreadGroup());     }          public DefaultThreadFactory(String poolName, boolean daemon, int priority, ThreadGroup threadGroup) {         ...         //prefix用来标记线程名字的前缀         prefix = poolName + '-' + poolId.incrementAndGet() + '-';         this.daemon = daemon;         this.priority = priority;         this.threadGroup = threadGroup;     }          @Override     public Thread newThread(Runnable r) {         Thread t = newThread(new DefaultRunnableDecorator(r), prefix + nextId.incrementAndGet());         if (t.isDaemon()) {             if (!daemon) t.setDaemon(false);         } else {             if (daemon) t.setDaemon(true);         }         if (t.getPriority() != priority) t.setPriority(priority);         return t;     }          protected Thread newThread(Runnable r, String name) {         return new FastThreadLocalThread(threadGroup, r, name);     }     ... }

NioEventLoop是如何与一个线程实体绑定的？NioEventLoop会通过线程执行器ThreadPerTaskExecutor创建一个FastThreadLocalThread，然后再将该FastThreadLocalThread线程保存到其成员变量中，从而实现与一个线程实体进行绑定。

 

(4)NioEventLoop的启动总结

一.在注册服务端Channel的过程中，主线程最终会调用AbstractUnsafe的register()方法。该方法首先会将一个NioEventLoop绑定到这个服务端Channel上，然后把实际注册Selector的逻辑封装成一个Runnable任务，接着调用NioEventLoop的execute()方法来执行这个Runnable任务。

 

二.NioEventLoop的execute()方法其实就是其父类SingleThreadEventExecutor的execute()方法，它会先判断当前调用execute()方法的线程是不是Netty的Reactor线程，如果不是就调用startThread()方法来创建一个Reactor线程。

 

三.startThread()方法会通过线程执行器ThreadPerTaskExecutor的execute()方法来创建一个线程。这个线程是一个FastThreadLocalThread线程，这个线程需要执行如下逻辑：把线程保存到NioEventLoop的成员变量thread中，然后调用NioEventLoop的run()方法来启动NioEventLoop。

 

NioEventLoop的启动流程如下：

bind() -> initAndRegister() -> config().group().register() -> eventloop.execute() //入口   startThread() -> doStartThread() //创建线程     ThreadPerTaskExecutor.execute() //线程执行器创建FastThreadLocalThread线程       thread = Thread.currentThread() //保存FastThreadLocalThread线程到NioEventLoop的成员变量中       NioEventLoop.run() //启动NioEventLoop

 

NioEventLoop的启动流程说明如下：

 

首先bind()方法会将具体绑定端口的操作封装成一个Runnable任务，然后调用NioEventLoop的execute()方法，接着Netty会判断调用execute()方法的线程是否是NIO线程，如果发现不是就会调用startThread()方法开始创建线程。

 

创建线程是通过线程执行器ThreadPerTaskExecutor来创建的。线程执行器的作用是每执行一个任务都会创建一个线程，而且创建出来的线程就是NioEventLoop底层的一个FastThreadLocalThread线程。

 

创建完FastThreadLocalThread线程后会执行一个Runnable任务，该Runnable任务首先会将这个线程保存到NioEventLoop对象。保存的目的是为了判断后续对NioEventLoop的相关执行线程是否为本身。如果不是就将封装好的一个任务放入TaskQueue中进行串行执行，实现线程安全。该Runnable任务然后会调用NioEventLoop的run()方法，从而启动NioEventLoop。NioEventLoop的run()方法是驱动Netty运转的核心方法。