• 解决Netty那些事儿之Reactor在Netty中的实现(创建篇)-上

    本系列Netty源码解析文章基于 4.1.56.Final版本

    在上篇文章深入讲解Netty那些事儿之从内核角度看IO模型中我们花了大量的篇幅来从内核角度详细讲述了五种IO模型的演进过程以及ReactorIO线程模型的底层基石IO多路复用技术在内核中的实现原理。

    最后我们引出了netty中使用的主从Reactor IO线程模型。

    netty中的reactor

    通过上篇文章的介绍，我们已经清楚了在IO调用的过程中内核帮我们搞了哪些事情，那么俗话说的好内核领进门，修行在netty，netty在用户空间又帮我们搞了哪些事情?

    那么从本文开始，笔者将从源码角度来带大家看下上图中的Reactor IO线程模型在Netty中是如何实现的。

    本文作为Reactor在Netty中实现系列文章中的开篇文章，笔者先来为大家介绍Reactor的骨架是如何创建出来的。

    在上篇文章中我们提到Netty采用的是主从Reactor多线程的模型，但是它在实现上又与Doug Lea在Scalable IO in Java论文中提到的经典主从Reactor多线程模型有所差异。

    经典主从Reactor多线程模型

    Netty中的Reactor是以Group的形式出现的，主从Reactor在Netty中就是主从Reactor组，每个Reactor Group中会有多个Reactor用来执行具体的IO任务。当然在netty中Reactor不只用来执行IO任务，这个我们后面再说。

    • Main Reactor Group中的Reactor数量取决于服务端要监听的端口个数，通常我们的服务端程序只会监听一个端口，所以Main Reactor Group只会有一个Main Reactor线程来处理最重要的事情：绑定端口地址，接收客户端连接，为客户端创建对应的SocketChannel，将客户端SocketChannel分配给一个固定的Sub Reactor。也就是上篇文章笔者为大家举的例子，饭店最重要的工作就是先把客人迎接进来。“我家大门常打开，开放怀抱等你，拥抱过就有了默契你会爱上这里......”

    • Sub Reactor Group里有多个Reactor线程，Reactor线程的个数可以通过系统参数-D io.netty.eventLoopThreads指定。默认的Reactor的个数为CPU核数 * 2。Sub Reactor线程主要用来轮询客户端SocketChannel上的IO就绪事件，处理IO就绪事件，执行异步任务。Sub Reactor Group做的事情就是上篇饭店例子中服务员的工作，客人进来了要为客人分配座位，端茶送水，做菜上菜。“不管远近都是客人，请不用客气，相约好了在一起，我们欢迎您......”

    一个客户端SocketChannel只能分配给一个固定的Sub Reactor。一个Sub Reactor负责处理多个客户端SocketChannel，这样可以将服务端承载的全量客户端连接分摊到多个Sub Reactor中处理，同时也能保证客户端SocketChannel上的IO处理的线程安全性。

    由于文章篇幅的关系，作为Reactor在netty中实现的第一篇我们主要来介绍主从Reactor Group的创建流程，骨架脉络先搭好。

    下面我们来看一段Netty服务端代码的编写模板，从代码模板的流程中我们来解析下主从Reactor的创建流程以及在这个过程中所涉及到的Netty核心类。

     资料直通车：Linux内核源码技术学习路线+视频教程内核源码

    学习直通车：Linux内核源码内存调优文件系统进程管理设备驱动/网络协议栈

    Netty服务端代码模板

    1. /**
    2.  * Echoes back any received data from a client.
    3.  */
    4. public final class EchoServer {
    5.     static final int PORT = Integer.parseInt(System.getProperty("port", "8007"));
    6.  
    7.     public static void main(String[] args) throws Exception {
    8.         // Configure the server.
    9.         //创建主从Reactor线程组
    10.         EventLoopGroup bossGroup = new NioEventLoopGroup(1);
    11.         EventLoopGroup workerGroup = new NioEventLoopGroup();
    12.         final EchoServerHandler serverHandler = new EchoServerHandler();
    13.         try {
    14.             ServerBootstrap b = new ServerBootstrap();
    15.             b.group(bossGroup, workerGroup)//配置主从Reactor
    16.              .channel(NioServerSocketChannel.class)//配置主Reactor中的channel类型
    17.              .option(ChannelOption.SO_BACKLOG, 100)//设置主Reactor中channel的option选项
    18.              .handler(new LoggingHandler(LogLevel.INFO))//设置主Reactor中Channel->pipline->handler
    19.              .childHandler(new ChannelInitializer<SocketChannel>() {//设置从Reactor中注册channel的pipeline
    20.                  @Override
    21.                  public void initChannel(SocketChannel ch) throws Exception {
    22.                      ChannelPipeline p = ch.pipeline();
    23.                      //p.addLast(new LoggingHandler(LogLevel.INFO));
    24.                      p.addLast(serverHandler);
    25.                  }
    26.              });
    27.  
    28.             // Start the server. 绑定端口启动服务，开始监听accept事件
    29.             ChannelFuture f = b.bind(PORT).sync();
    30.             // Wait until the server socket is closed.
    31.             f.channel().closeFuture().sync();
    32.         } finally {
    33.             // Shut down all event loops to terminate all threads.
    34.             bossGroup.shutdownGracefully();
    35.             workerGroup.shutdownGracefully();
    36.         }
    37.     }
    38. }
    39.  
    40.

    1. 首先我们要创建Netty最核心的部分 -> 创建主从Reactor Group，在Netty中EventLoopGroup就是Reactor Group的实现类。对应的EventLoop就是Reactor的实现类。

    1.   //创建主从Reactor线程组
    2.   EventLoopGroup bossGroup = new NioEventLoopGroup(1);
    3.   EventLoopGroup workerGroup = new NioEventLoopGroup();

    1. 创建用于IO处理的ChannelHandler，实现相应IO事件的回调函数，编写对应的IO处理逻辑。注意这里只是简单示例哈，详细的IO事件处理，笔者会单独开一篇文章专门讲述。

    1. final EchoServerHandler serverHandler = new EchoServerHandler();
    2.  
    3. /**
    4.  * Handler implementation for the echo server.
    5.  */
    6. @Sharable
    7. public class EchoServerHandler extends ChannelInboundHandlerAdapter {
    8.  
    9.     @Override
    10.     public void channelRead(ChannelHandlerContext ctx, Object msg) {
    11.         ................省略IO处理逻辑................
    12.         ctx.write(msg);
    13.     }
    14.  
    15.     @Override
    16.     public void channelReadComplete(ChannelHandlerContext ctx) {
    17.         
    18.         ctx.flush();
    19.     }
    20.  
    21.     @Override
    22.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
    23.         // Close the connection when an exception is raised.
    24.         cause.printStackTrace();
    25.         ctx.close();
    26.     }
    27. }

    1. 创建ServerBootstrapNetty服务端启动类，并在启动类中配置启动Netty服务端所需要的一些必备信息。

      在上篇文章介绍Socket内核结构小节中我们提到，在编写服务端网络程序时，我们首先要创建一个Socket用于listen和bind端口地址，我们把这个叫做监听Socket,这里对应的就是NioServerSocketChannel.class。当客户端连接完成三次握手，系统调用accept函数会基于监听Socket创建出来一个新的Socket专门用于与客户端之间的网络通信我们称为客户端连接Socket,这里对应的就是NioSocketChannel.class

      netty有两种Channel类型：一种是服务端用于监听绑定端口地址的NioServerSocketChannel,一种是用于客户端通信的NioSocketChannel。每种Channel类型实例都会对应一个PipeLine用于编排对应channel实例上的IO事件处理逻辑。PipeLine中组织的就是ChannelHandler用于编写特定的IO处理逻辑。

      注意serverBootstrap.handler设置的是服务端NioServerSocketChannel PipeLine中的ChannelHandler。

      ServerBootstrap启动类方法带有child前缀的均是设置客户端NioSocketChannel属性的。

      ChannelInitializer是用于当SocketChannel成功注册到绑定的Reactor上后，用于初始化该SocketChannel的Pipeline。它的initChannel方法会在注册成功后执行。这里只是捎带提一下，让大家有个初步印象，后面我会专门介绍。

      • serverBootstrap.childHandler(ChannelHandler childHandler)用于设置客户端NioSocketChannel中对应Pipieline中的ChannelHandler。我们通常配置的编码解码器就是在这里。

      • serverBootstrap.option(ChannelOption.SO_BACKLOG, 100)设置服务端ServerSocketChannel中的SocketOption。关于SocketOption的选项我们后边的文章再聊，本文主要聚焦在Netty Main Reactor Group的创建及工作流程。

      • serverBootstrap.handler(....)设置服务端NioServerSocketChannel中对应Pipieline中的ChannelHandler。

      • 通过serverBootstrap.group(bossGroup, workerGroup)为Netty服务端配置主从Reactor Group实例。

      • 通过serverBootstrap.channel(NioServerSocketChannel.class)配置Netty服务端的ServerSocketChannel用于绑定端口地址以及创建客户端SocketChannel。Netty中的NioServerSocketChannel.class就是对JDK NIO中ServerSocketChannel的封装。而用于表示客户端连接的NioSocketChannel是对JDK NIO SocketChannel封装。

    2. ChannelFuture f = serverBootstrap.bind(PORT).sync()这一步会是下篇文章要重点分析的主题Main Reactor Group的启动，绑定端口地址，开始监听客户端连接事件（OP_ACCEPT）。本文我们只关注创建流程。

    3. f.channel().closeFuture().sync()等待服务端NioServerSocketChannel关闭。Netty服务端到这里正式启动，并准备好接受客户端连接的准备。

    4. shutdownGracefully优雅关闭主从Reactor线程组里的所有Reactor线程。

    Netty对IO模型的支持

    在上篇文章中我们介绍了五种IO模型，Netty中支持BIO,NIO,AIO以及多种操作系统下的IO多路复用技术实现。

    在Netty中切换这几种IO模型也是非常的方便，下面我们来看下Netty如何对这几种IO模型进行支持的。

    首先我们介绍下几个与IO模型相关的重要接口：

    EventLoop

    EventLoop就是Netty中的Reactor，可以说它就是Netty的引擎，负责Channel上IO就绪事件的监听，IO就绪事件的处理，异步任务的执行驱动着整个Netty的运转。

    不同IO模型下，EventLoop有着不同的实现，我们只需要切换不同的实现类就可以完成对NettyIO模型的切换。

    BIONIOAIO
    ThreadPerChannelEventLoopNioEventLoopAioEventLoop

    在NIO模型下Netty会自动根据操作系统以及版本的不同选择对应的IO多路复用技术实现。比如Linux 2.6版本以上用的是Epoll，2.6版本以下用的是Poll，Mac下采用的是Kqueue。

    其中Linux kernel 在5.1版本引入的异步IO库io_uring正在netty中孵化。

    EventLoopGroup

    Netty中的Reactor是以Group的形式出现的，EventLoopGroup正是Reactor组的接口定义，负责管理Reactor，Netty中的Channel就是通过EventLoopGroup注册到具体的Reactor上的。

    Netty的IO线程模型是主从Reactor多线程模型，主从Reactor线程组在Netty源码中对应的其实就是两个EventLoopGroup实例。

    不同的IO模型也有对应的实现：

    BIONIOAIO
    ThreadPerChannelEventLoopGroupNioEventLoopGroupAioEventLoopGroup

    ServerSocketChannel

    用于Netty服务端使用的ServerSocketChannel，对应于上篇文章提到的监听Socket，负责绑定监听端口地址，接收客户端连接并创建用于与客户端通信的SocketChannel。

    不同的IO模型下的实现：

    BIONIOAIO
    OioServerSocketChannelNioServerSocketChannelAioServerSocketChannel

    SocketChannel

    用于与客户端通信的SocketChannel，对应于上篇文章提到的客户端连接Socket，当客户端完成三次握手后，由系统调用accept函数根据监听Socket创建。

    不同的IO模型下的实现：

    BIONIOAIO
    OioSocketChannelNioSocketChannelAioSocketChannel

    我们看到在不同IO模型的实现中，Netty这些围绕IO模型的核心类只是前缀的不同：

    • BIO对应的前缀为Oio表示old io，现在已经废弃不推荐使用。

    • NIO对应的前缀为Nio，正是Netty推荐也是我们常用的非阻塞IO模型。

    • AIO对应的前缀为Aio，由于Linux下的异步IO机制实现的并不成熟，性能提升表现上也不明显，现已被删除。

    我们只需要将IO模型的这些核心接口对应的实现类前缀改为对应IO模型的前缀，就可以轻松在Netty中完成对IO模型的切换。

    image.png

    多种NIO的实现

    CommonLinuxMac
    NioEventLoopGroupEpollEventLoopGroupKQueueEventLoopGroup
    NioEventLoopEpollEventLoopKQueueEventLoop
    NioServerSocketChannelEpollServerSocketChannelKQueueServerSocketChannel
    NioSocketChannelEpollSocketChannelKQueueSocketChannel

    我们通常在使用NIO模型的时候会使用Common列下的这些IO模型核心类，Common类也会根据操作系统的不同自动选择JDK在对应平台下的IO多路复用技术的实现。

    而Netty自身也根据操作系统的不同提供了自己对IO多路复用技术的实现，比JDK的实现性能更优。比如：

    • JDK的 NIO 默认实现是水平触发，Netty 是边缘触发(默认)和水平触发可切换。。

    • Netty 实现的垃圾回收更少、性能更好。

    我们编写Netty服务端程序的时候也可以根据操作系统的不同，采用Netty自身的实现来进一步优化程序。做法也很简单，直接将上图中红框里的实现类替换成Netty的自身实现类即可完成切换。

    经过以上对Netty服务端代码编写模板以及IO模型相关核心类的简单介绍，我们对Netty的创建流程有了一个简单粗略的总体认识，下面我们来深入剖析下创建流程过程中的每一个步骤以及这个过程中涉及到的核心类实现。

    以下源码解析部分我们均采用Common列下NIO相关的实现进行解析。

    创建主从Reactor线程组

    在Netty服务端程序编写模板的开始，我们首先会创建两个Reactor线程组：

    netty中的reactor.png

    • 一个是主Reactor线程组bossGroup用于监听客户端连接，创建客户端连接NioSocketChannel，并将创建好的客户端连接NioSocketChannel注册到从Reactor线程组中一个固定的Reactor上。

    • 一个是从Reactor线程组workerGroup，workerGroup中的Reactor负责监听绑定在其上的客户端连接NioSocketChannel上的IO就绪事件，并处理IO就绪事件，执行异步任务。

    1.   //创建主从Reactor线程组
    2.   EventLoopGroup bossGroup = new NioEventLoopGroup(1);
    3.   EventLoopGroup workerGroup = new NioEventLoopGroup();

    Netty中Reactor线程组的实现类为NioEventLoopGroup，在创建bossGroup和workerGroup的时候用到了NioEventLoopGroup的两个构造函数：

    • 带nThreads参数的构造函数public NioEventLoopGroup(int nThreads)。

    • 不带nThreads参数的默认构造函数public NioEventLoopGroup()

    1. public class NioEventLoopGroup extends MultithreadEventLoopGroup {
    2.  
    3.     /**
    4.      * Create a new instance using the default number of threads, the default {@link ThreadFactory} and
    5.      * the {@link SelectorProvider} which is returned by {@link SelectorProvider#provider()}.
    6.      */
    7.     public NioEventLoopGroup() {
    8.         this(0);
    9.     }
    10.  
    11.     /**
    12.      * Create a new instance using the specified number of threads, {@link ThreadFactory} and the
    13.      * {@link SelectorProvider} which is returned by {@link SelectorProvider#provider()}.
    14.      */
    15.     public NioEventLoopGroup(int nThreads) {
    16.         this(nThreads, (Executor) null);
    17.     }
    18.  
    19.     ......................省略...........................
    20. }

    nThreads参数表示当前要创建的Reactor线程组内包含多少个Reactor线程。不指定nThreads参数的话采用默认的Reactor线程个数，用0表示。

    最终会调用到构造函数

    1.     public NioEventLoopGroup(int nThreads, Executor executor, final SelectorProvider selectorProvider,
    2.                              final SelectStrategyFactory selectStrategyFactory) {
    3.         super(nThreads, executor, selectorProvider, selectStrategyFactory, RejectedExecutionHandlers.reject());
    4.     }

    下面简单介绍下构造函数中这几个参数的作用，后面我们在讲解本文主线的过程中还会提及这几个参数，到时在详细介绍，这里只是让大家有个初步印象，不必做过多的纠缠。

    • Executor executor：负责启动Reactor线程进而Reactor才可以开始工作。

    Reactor线程组NioEventLoopGroup负责创建Reactor线程，在创建的时候会将executor传入。

    • RejectedExecutionHandler： 当向Reactor添加异步任务添加失败时，采用的拒绝策略。Reactor的任务不只是监听IO活跃事件和IO任务的处理，还包括对异步任务的处理。这里大家只需有个这样的概念，后面笔者会专门详细介绍。

    • SelectorProvider selectorProvider： Reactor中的IO模型为IO多路复用模型，对应于JDK NIO中的实现为java.nio.channels.Selector（就是我们上篇文章中提到的select,poll,epoll），每个Reator中都包含一个Selector，用于轮询注册在该Reactor上的所有Channel上的IO事件。SelectorProvider就是用来创建Selector的。

    • SelectStrategyFactory selectStrategyFactory： Reactor最重要的事情就是轮询注册其上的Channel上的IO就绪事件，这里的SelectStrategyFactory用于指定轮询策略，默认为DefaultSelectStrategyFactory.INSTANCE。

    最终会将这些参数交给NioEventLoopGroup的父类构造器，下面我们来看下NioEventLoopGroup类的继承结构：

    image.png

    NioEventLoopGroup类的继承结构乍一看比较复杂，大家不要慌，笔者会随着主线的深入慢慢地介绍这些父类接口，我们现在重点关注Mutithread前缀的类。

    我们知道NioEventLoopGroup是Netty中的Reactor线程组的实现，既然是线程组那么肯定是负责管理和创建多个Reactor线程的，所以Mutithread前缀的类定义的行为自然是对Reactor线程组内多个Reactor线程的创建和管理工作。

    MultithreadEventLoopGroup

    1. public abstract class MultithreadEventLoopGroup extends MultithreadEventExecutorGroup implements EventLoopGroup {
    2.  
    3.     private static final InternalLogger logger = InternalLoggerFactory.getInstance(MultithreadEventLoopGroup.class);
    4.     //默认Reactor个数
    5.     private static final int DEFAULT_EVENT_LOOP_THREADS;
    6.  
    7.     static {
    8.         DEFAULT_EVENT_LOOP_THREADS = Math.max(1, SystemPropertyUtil.getInt(
    9.                 "io.netty.eventLoopThreads", NettyRuntime.availableProcessors() * 2));
    10.  
    11.         if (logger.isDebugEnabled()) {
    12.             logger.debug("-Dio.netty.eventLoopThreads: {}", DEFAULT_EVENT_LOOP_THREADS);
    13.         }
    14.     }
    15.  
    16.     /**
    17.      * @see MultithreadEventExecutorGroup#MultithreadEventExecutorGroup(int, Executor, Object...)
    18.      */
    19.     protected MultithreadEventLoopGroup(int nThreads, Executor executor, Object... args) {
    20.         super(nThreads == 0 ? DEFAULT_EVENT_LOOP_THREADS : nThreads, executor, args);
    21.     }
    22.  
    23.     ...................省略.....................
    24. }

    MultithreadEventLoopGroup类主要的功能就是用来确定Reactor线程组内Reactor的个数。

    默认的Reactor的个数存放于字段DEFAULT_EVENT_LOOP_THREADS中。

    从static {}静态代码块中我们可以看出默认Reactor的个数的获取逻辑：

    • 可以通过系统变量 -D io.netty.eventLoopThreads"指定。

    • 如果不指定，那么默认的就是NettyRuntime.availableProcessors() * 2

    当nThread参数设置为0采用默认设置时，Reactor线程组内的Reactor个数则设置为DEFAULT_EVENT_LOOP_THREADS。

    MultithreadEventExecutorGroup

    MultithreadEventExecutorGroup这里就是本小节的核心，主要用来定义创建和管理Reactor的行为。

    1. public abstract class MultithreadEventExecutorGroup extends AbstractEventExecutorGroup {
    2.  
    3.     //Reactor线程组中的Reactor集合
    4.     private final EventExecutor[] children;
    5.     private final Set<EventExecutor> readonlyChildren;
    6.     //从Reactor group中选择一个特定的Reactor的选择策略 用于channel注册绑定到一个固定的Reactor上
    7.     private final EventExecutorChooserFactory.EventExecutorChooser chooser;
    8.  
    9.     /**
    10.      * Create a new instance.
    11.      *
    12.      * @param nThreads          the number of threads that will be used by this instance.
    13.      * @param executor          the Executor to use, or {@code null} if the default should be used.
    14.      * @param args              arguments which will passed to each {@link #newChild(Executor, Object...)} call
    15.      */
    16.     protected MultithreadEventExecutorGroup(int nThreads, Executor executor, Object... args) {
    17.         this(nThreads, executor, DefaultEventExecutorChooserFactory.INSTANCE, args);
    18.     }
    19.  
    20.     ............................省略................................
    21. }

    首先介绍一个新的构造器参数EventExecutorChooserFactory chooserFactory。当客户端连接完成三次握手后，Main Reactor会创建客户端连接NioSocketChannel，并将其绑定到Sub Reactor Group中的一个固定Reactor，那么具体要绑定到哪个具体的Sub Reactor上呢？这个绑定策略就是由chooserFactory来创建的。默认为DefaultEventExecutorChooserFactory。

    下面就是本小节的主题Reactor线程组的创建过程：

    1.     protected MultithreadEventExecutorGroup(int nThreads, Executor executor,
    2.                                             EventExecutorChooserFactory chooserFactory, Object... args) {
    3.         if (nThreads <= 0) {
    4.             throw new IllegalArgumentException(String.format("nThreads: %d (expected: > 0)", nThreads));
    5.         }
    6.  
    7.         if (executor == null) {
    8.             //用于创建Reactor线程
    9.             executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());
    10.         }
    11.  
    12.         children = new EventExecutor[nThreads];
    13.         //循环创建reaactor group中的Reactor
    14.         for (int i = 0; i < nThreads; i ++) {
    15.             boolean success = false;
    16.             try {
    17.                 //创建reactor
    18.                 children[i] = newChild(executor, args);
    19.                 success = true;
    20.             } catch (Exception e) {
    21.                 throw new IllegalStateException("failed to create a child event loop", e);
    22.             } finally {
    23.                      ................省略................
    24.                 }
    25.             }
    26.         }
    27.         //创建channel到Reactor的绑定策略
    28.         chooser = chooserFactory.newChooser(children);
    29.  
    30.          ................省略................
    31.  
    32.         Set<EventExecutor> childrenSet = new LinkedHashSet<EventExecutor>(children.length);
    33.         Collections.addAll(childrenSet, children);
    34.         readonlyChildren = Collections.unmodifiableSet(childrenSet);
    35.     }

    1. 创建用于启动Reactor线程的executor

    在Netty Reactor Group中的单个Reactor的IO线程模型为上篇文章提到的单Reactor单线程模型，一个Reactor线程负责轮询注册其上的所有Channel中的IO就绪事件，处理IO事件，执行Netty中的异步任务等工作。正是这个Reactor线程驱动着整个Netty的运转，可谓是Netty的核心引擎。

    单Reactor单线程模型.png

    而这里的executor就是负责启动Reactor线程的，从创建源码中我们可以看到executor的类型为ThreadPerTaskExecutor。

    ThreadPerTaskExecutor

    1. public final class ThreadPerTaskExecutor implements Executor {
    2.     private final ThreadFactory threadFactory;
    3.  
    4.     public ThreadPerTaskExecutor(ThreadFactory threadFactory) {
    5.         this.threadFactory = ObjectUtil.checkNotNull(threadFactory, "threadFactory");
    6.     }
    7.  
    8.     @Override
    9.     public void execute(Runnable command) {
    10.         threadFactory.newThread(command).start();
    11.     }
    12. }

    我们看到ThreadPerTaskExecutor做的事情很简单，从它的命名前缀ThreadPerTask我们就可以猜出它的工作方式，就是来一个任务就创建一个线程执行。而创建的这个线程正是netty的核心引擎Reactor线程。

    在Reactor线程启动的时候，Netty会将Reactor线程要做的事情封装成Runnable，丢给exexutor启动。

    而Reactor线程的核心就是一个死循环不停的轮询IO就绪事件，处理IO事件，执行异步任务。一刻也不停歇，堪称996典范。

    这里向大家先卖个关子，"Reactor线程是何时启动的呢？？"

    2. 创建Reactor

    Reactor线程组NioEventLoopGroup包含多个Reactor，存放于private final EventExecutor[] children数组中。

    所以下面的事情就是创建nThread个Reactor，并存放于EventExecutor[] children字段中，

    我们来看下用于创建Reactor的newChild(executor, args)方法：

    newChild

    newChild方法是MultithreadEventExecutorGroup中的一个抽象方法，提供给具体子类实现。

    protected abstract EventExecutor newChild(Executor executor, Object... args) throws Exception;

    这里我们解析的是NioEventLoopGroup，我们来看下newChild在该类中的实现：

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

    前边提到的众多构造器参数，这里会通过可变参数Object... args传入到Reactor类NioEventLoop的构造器中。

    这里介绍下新的参数EventLoopTaskQueueFactory queueFactory，前边提到Netty中的Reactor主要工作是轮询注册其上的所有Channel上的IO就绪事件，处理IO就绪事件。除了这些主要的工作外，Netty为了极致的压榨Reactor的性能，还会让它做一些异步任务的执行工作。既然要执行异步任务，那么Reactor中就需要一个队列来保存任务。

    这里的EventLoopTaskQueueFactory就是用来创建这样的一个队列来保存Reactor中待执行的异步任务。

    可以把Reactor理解成为一个单线程的线程池，类似于JDK中的SingleThreadExecutor，仅用一个线程来执行轮询IO就绪事件，处理IO就绪事件，执行异步任务。同时待执行的异步任务保存在Reactor里的taskQueue中。

    NioEventLoop

    1. public final class NioEventLoop extends SingleThreadEventLoop {
    2.     //用于创建JDK NIO Selector,ServerSocketChannel
    3.     private final SelectorProvider provider;
    4.     //Selector轮询策略 决定什么时候轮询，什么时候处理IO事件，什么时候执行异步任务
    5.     private final SelectStrategy selectStrategy;
    6.     /**
    7.      * The NIO {@link Selector}.
    8.      */
    9.     private Selector selector;
    10.     private Selector unwrappedSelector;
    11.  
    12.     NioEventLoop(NioEventLoopGroup parent, Executor executor, SelectorProvider selectorProvider,
    13.                  SelectStrategy strategy, RejectedExecutionHandler rejectedExecutionHandler,
    14.                  EventLoopTaskQueueFactory queueFactory) {
    15.         super(parent, executor, false, newTaskQueue(queueFactory), newTaskQueue(queueFactory),
    16.                 rejectedExecutionHandler);
    17.         this.provider = ObjectUtil.checkNotNull(selectorProvider, "selectorProvider");
    18.         this.selectStrategy = ObjectUtil.checkNotNull(strategy, "selectStrategy");
    19.         final SelectorTuple selectorTuple = openSelector();
    20.         this.selector = selectorTuple.selector;
    21.         this.unwrappedSelector = selectorTuple.unwrappedSelector;
    22.     }
    23. }

    这里就正式开始了Reactor的创建过程，我们知道Reactor的核心是采用的IO多路复用模型来对客户端连接上的IO事件进行监听，所以最重要的事情是创建Selector(JDK NIO 中IO多路复用技术的实现)。

    可以把Selector理解为我们上篇文章介绍的Select,poll,epoll，它是JDK NIO对操作系统内核提供的这些IO多路复用技术的封装。

    openSelector

    openSelector是NioEventLoop类中用于创建IO多路复用的Selector，并对创建出来的JDK NIO 原生的Selector进行性能优化。

    首先会通过SelectorProvider#openSelector创建JDK NIO原生的Selector。

    1.  private SelectorTuple openSelector() {
    2.         final Selector unwrappedSelector;
    3.         try {
    4.             //通过JDK NIO SelectorProvider创建Selector
    5.             unwrappedSelector = provider.openSelector();
    6.         } catch (IOException e) {
    7.             throw new ChannelException("failed to open a new selector", e);
    8.         }
    9.  
    10.         ..................省略.............
    11. }

    SelectorProvider会根据操作系统的不同选择JDK在不同操作系统版本下的对应Selector的实现。Linux下会选择Epoll，Mac下会选择Kqueue。

    下面我们就来看下SelectorProvider是如何做到自动适配不同操作系统下IO多路复用实现的

    SelectorProvider

    1.     public NioEventLoopGroup(ThreadFactory threadFactory) {
    2.         this(0, threadFactory, SelectorProvider.provider());
    3.     }

    SelectorProvider是在前面介绍的NioEventLoopGroup类构造函数中通过调用SelectorProvider.provider()被加载，并通过NioEventLoopGroup#newChild方法中的可变长参数Object... args传递到NioEventLoop中的private final SelectorProvider provider字段中。

    SelectorProvider的加载过程：

    1. public abstract class SelectorProvider {
    2.  
    3.     public static SelectorProvider provider() {
    4.         synchronized (lock) {
    5.             if (provider != null)
    6.                 return provider;
    7.             return AccessController.doPrivileged(
    8.                 new PrivilegedAction() {
    9.                     public SelectorProvider run() {
    10.                             if (loadProviderFromProperty())
    11.                                 return provider;
    12.                             if (loadProviderAsService())
    13.                                 return provider;
    14.                             provider = sun.nio.ch.DefaultSelectorProvider.create();
    15.                             return provider;
    16.                         }
    17.                     });
    18.         }
    19.     }
    20. }

    从SelectorProvider加载源码中我们可以看出，SelectorProvider的加载方式有三种，优先级如下：

    1. 通过系统变量-D java.nio.channels.spi.SelectorProvider指定SelectorProvider的自定义实现类全限定名。通过应用程序类加载器(Application Classloader)加载。

    1.     private static boolean loadProviderFromProperty() {
    2.         String cn = System.getProperty("java.nio.channels.spi.SelectorProvider");
    3.         if (cn == null)
    4.             return false;
    5.         try {
    6.             Class<?> c = Class.forName(cn, true,
    7.                                        ClassLoader.getSystemClassLoader());
    8.             provider = (SelectorProvider)c.newInstance();
    9.             return true;
    10.         } 
    11.         .................省略.............
    12.     }

    1. 通过SPI方式加载。在工程目录META-INF/services下定义名为java.nio.channels.spi.SelectorProvider的SPI文件，文件中第一个定义的SelectorProvider实现类全限定名就会被加载。

    1.     private static boolean loadProviderAsService() {
    2.  
    3.         ServiceLoader<SelectorProvider> sl =
    4.             ServiceLoader.load(SelectorProvider.class,
    5.                                ClassLoader.getSystemClassLoader());
    6.         Iterator<SelectorProvider> i = sl.iterator();
    7.         for (;;) {
    8.             try {
    9.                 if (!i.hasNext())
    10.                     return false;
    11.                 provider = i.next();
    12.                 return true;
    13.             } catch (ServiceConfigurationError sce) {
    14.                 if (sce.getCause() instanceof SecurityException) {
    15.                     // Ignore the security exception, try the next provider
    16.                     continue;
    17.                 }
    18.                 throw sce;
    19.             }
    20.         }
    21.     }

    1. 如果以上两种方式均未被定义，那么就采用SelectorProvider系统默认实现sun.nio.ch.DefaultSelectorProvider。笔者当前使用的操作系统是MacOS，从源码中我们可以看到自动适配了KQueue实现。

    1. public class DefaultSelectorProvider {
    2.     private DefaultSelectorProvider() {
    3.     }
    4.  
    5.     public static SelectorProvider create() {
    6.         return new KQueueSelectorProvider();
    7.     }
    8. }

    不同操作系统中JDK对于DefaultSelectorProvider会有所不同，Linux内核版本2.6以上对应的Epoll，Linux内核版本2.6以下对应的Poll，MacOS对应的是KQueue。

    下面我们接着回到io.netty.channel.nio.NioEventLoop#openSelector的主线上来。

     

    文章篇幅有限 下文讲解

  • 相关阅读:
    Redis常见命令
    Vue3+Echarts: 浏览器缩小后，图表内容发生重叠
    六要素一体微型气象站介绍
    Python数据挖掘：入门、进阶与实用案例分析——自动售货机销售数据分析与应用
    HAProxy 1-9-0-porting-guide
    微信小程序 doc
    matplotlib
    C Primer Plus(6) 中文版 第14章 结构和其他数据形式 14.1 示例问题：创建图书目录
    本地库覆盖glibc的API
    MybatisPlus搭建项目
  • 原文地址：https://blog.csdn.net/youzhangjing_/article/details/128150895