【对话写 Netty 代码的同学，你真的懂 Netty 了吗？（二）】主线流程 new NioEventLoopGroup(nThreads) 究竟做了什么

比眉伴天荒 2022-11-19 01:13 160阅读 0赞

### 写在前面 ###

本篇开始之前，我们先回顾一下上篇的内容：[对话写 Netty 代码的同学，你真的懂 Netty 了吗？（一）][Netty _ Netty]

我们在第一篇中带大家简单了解了一下 Netty 源码中的核心类文件。  
大类包含：

> 门面 - 引导器 以 AbstractBootstrap 为代表  
> 传输管道 以 ChannelPipeline 为代表  
> Netty 自定义线程池 以 EventLoop 为代表  
> 通道 以 AbstractChannel 为代表  
> 上下文 以 AbstractChannelHandlerContext 为代表

### 目录： ###

[【对话写 Netty 代码的同学，你真的懂 Netty 了吗？（一）】Netty 初始化总览][Netty _ Netty]

[【对话写 Netty 代码的同学，你真的懂 Netty 了吗？（二）】主线流程 new NioEventLoopGroup(nThreads) 究竟做了什么][Netty _ Netty _ new NioEventLoopGroup_nThreads_]

[【对话写 Netty 代码的同学，你真的懂 Netty 了吗？（三上）】initAndRegister() 之 init() 怒了！这文章为啥写成了这样？？][Netty _ Netty _initAndRegister_ _ init_]

[【对话写 Netty 代码的同学，你真的懂 Netty 了吗？（三中）】initAndRegister() 之 init() 对不起！！我柜子动了…我不学了][Netty _ Netty _initAndRegister_ _ init_ 1]

### Netty 主线流程 bootstrap.bind(port) ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3c5MDMzMjg2MTU_size_16_color_FFFFFF_t_70]

public class NettyServer { 
        public static void main(String[] args) { 
            EventLoopGroup bossGroup = new NioEventLoopGroup(1);
            EventLoopGroup workerGroup = new NioEventLoopGroup(0);
            try { 
                ServerBootstrap bootstrap = new ServerBootstrap();
                bootstrap.group(bossGroup, workerGroup)
                        .channel(NioServerSocketChannel.class)
                        .option(ChannelOption.SO_BACKLOG, 1024)
                        .childHandler(new ChannelInitializer<SocketChannel>() { 
                            @Override
                            protected void initChannel(SocketChannel socketChannel) throws Exception { 
                             socketChannel.pipeline().addLast(new MyNettyServerHandler());
                            }
                        });
                ChannelFuture cf = bootstrap.bind(9000).sync();
                cf.channel().closeFuture().sync();
    
            } catch (Exception e) { 
                e.printStackTrace();
            } finally { 
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }
    
    }

这里我贴出来 Netty 架构图，和样例模板代码。  
其实 Netty 的主线流程我们只需要关注 ，三行核心代码即可：

new NioEventLoopGroup(1);
    new ServerBootstrap();
    bootstrap.bind(9000).sync();

其他都是边角的赋值代码，其中最为核心的是：

bootstrap.bind(9000).sync();

那么本篇将围绕上述三行代码进行展开

#### 第一部分 new NioEventLoopGroup(nThreads) ####

在写 Netty 程序，必然我们要声明 bossGroup 和 workerGroup  
那首先来看看 new NioEventLoopGroup(nThreads) 到底为我们做了什么

其继承结构如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3c5MDMzMjg2MTU_size_16_color_FFFFFF_t_70 1]

`NioEventLoopGroup` 构造方法：

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) { 
            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());
        }

经过一系列的重载，最后调用到：父类的构造方法 MultithreadEventLoopGroup

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

继续向上 MultithreadEventExecutorGroup

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

最终真正的执行逻辑在 MultithreadEventExecutorGroup 类中

> Create a new instance.  
> Params:  
> nThreads – the number of threads that will be used by this instance.  
> executor – the Executor to use, or null if the default should be used.  
> chooserFactory – the EventExecutorChooserFactory to use.  
> args – arguments which will passed to each newChild(Executor, Object…) call

> 创建新实例。  
> 参数：  
> nThreads–此实例将使用的线程数。  
> executor–要使用的executor，如果应该使用默认值，则为null。  
> chooserFactory–要使用的事件执行器chooserFactory。  
> args–将传递给每个newChild（Executor、Object…）调用的参数

protected MultithreadEventExecutorGroup(int nThreads, Executor executor,
                                                EventExecutorChooserFactory chooserFactory, Object... args) { 
            if (nThreads <= 0) { 
                throw new IllegalArgumentException(String.format("nThreads: %d (expected: > 0)", nThreads));
            }
    
            if (executor == null) { 
                executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());
            }
    
            children = new EventExecutor[nThreads];
    
            for (int i = 0; i < nThreads; i ++) { 
                boolean success = false;
                try { 
                    children[i] = newChild(executor, args);
                    success = true;
                } catch (Exception e) { 
                    // TODO: Think about if this is a good exception type
                    throw new IllegalStateException("failed to create a child event loop", e);
                } finally { 
                    if (!success) { 
                        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;
                            }
                        }
                    }
                }
            }
    
            chooser = chooserFactory.newChooser(children);
    
            final FutureListener<Object> terminationListener = new FutureListener<Object>() { 
                @Override
                public void operationComplete(Future<Object> future) throws Exception { 
                    if (terminatedChildren.incrementAndGet() == children.length) { 
                        terminationFuture.setSuccess(null);
                    }
                }
            };
    
            for (EventExecutor e: children) { 
                e.terminationFuture().addListener(terminationListener);
            }
    
            Set<EventExecutor> childrenSet = new LinkedHashSet<EventExecutor>(children.length);
            Collections.addAll(childrenSet, children);
            readonlyChildren = Collections.unmodifiableSet(childrenSet);
        }

**重点看 children 的初始化代码：**

public abstract class MultithreadEventExecutorGroup extends AbstractEventExecutorGroup { 
    
    private final EventExecutor[] children;
        
    children = new EventExecutor[nThreads];
    children[i] = newChild(executor, args);    
    
    }

`children[i] = newChild(executor, args);` 很明显的 newChild 的方法这边调用的是 `NioEventLoopGroup` 的实现  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3c5MDMzMjg2MTU_size_16_color_FFFFFF_t_70 2]

@Override
        protected EventLoop newChild(Executor executor, Object... args) throws Exception { 
            return new NioEventLoop(this, executor, (SelectorProvider) args[0],
                ((SelectStrategyFactory) args[1]).newSelectStrategy(), (RejectedExecutionHandler) args[2]);
        }

public final class NioEventLoop extends SingleThreadEventLoop { 
        NioEventLoop(NioEventLoopGroup parent, Executor executor, SelectorProvider selectorProvider,
                     SelectStrategy strategy, RejectedExecutionHandler rejectedExecutionHandler) { 
            super(parent, executor, false, DEFAULT_MAX_PENDING_TASKS, rejectedExecutionHandler);
            if (selectorProvider == null) { 
                throw new NullPointerException("selectorProvider");
            }
            if (strategy == null) { 
                throw new NullPointerException("selectStrategy");
            }
            provider = selectorProvider;
            final SelectorTuple selectorTuple = openSelector();
            selector = selectorTuple.selector;
            unwrappedSelector = selectorTuple.unwrappedSelector;
            selectStrategy = strategy;
        }
    }

super 最终会调用到 `SingleThreadEventExecutor`

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");
            taskQueue = newTaskQueue(this.maxPendingTasks);
            rejectedExecutionHandler = ObjectUtil.checkNotNull(rejectedHandler, "rejectedHandler");
        }

这里有一行需要我们特别留意的代码 `taskQueue`：

private final Queue<Runnable> taskQueue;
    ……
    taskQueue = newTaskQueue(this.maxPendingTasks);

最后创建了一个 `LinkedBlockingQueue<Runnable>`

protected Queue<Runnable> newTaskQueue(int maxPendingTasks) { 
       return new LinkedBlockingQueue<Runnable>(maxPendingTasks);
    }

以及 Selecter 初始化代码：

/** * The NIO {@link Selector}. */
    private Selector selector;
    private Selector unwrappedSelector;
    private SelectedSelectionKeySet selectedKeys;
    private final SelectorProvider provider;
    
    provider = selectorProvider;
    final SelectorTuple selectorTuple = openSelector();
    selector = selectorTuple.selector;
    unwrappedSelector = selectorTuple.unwrappedSelector;
    selectStrategy = strategy;

结合上文给出的图和源码来看 MultithreadEventExecutorGroup.EventExecutor\[\] children 最终初始化的就是 NioEventLoop  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3c5MDMzMjg2MTU_size_16_color_FFFFFF_t_70 3]  
继承结构如下  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3c5MDMzMjg2MTU_size_16_color_FFFFFF_t_70 4]  
让我们重新一步步来完善这个架构图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3c5MDMzMjg2MTU_size_16_color_FFFFFF_t_70 5]  
Netty 作为一个异步的高性能框架，可以说内部使用了很多的线程去做事儿。包括我们看到的 bossGroup 、workGroup 其中都包含了很多不同的线程组，从现在看来，这两个 Group 所包含的 children 就是 NioEventLoop 的实例，其中每个 NioEventLoop 都包含一个 SingleThreadEvenetExecutor 线程实例。

注意这里，我们在创建 bossGroup 的时候一般传值为 1 即 `new NioEventLoopGroup(1);` 所以一般 bossGroup 只有一个 NioEventLoopGroup 线程处理

#### 第二部分 初始化赋值逻辑 ####

我们来看看，第二段代码 对 ServerBootstrap 进行了那些初始化赋值操作。  
这是一个链式编程的过程，其整体就是对我们 `new ServerBootstrap()`这个实例进行一个初始化的赋值。

ServerBootstrap bootstrap = new ServerBootstrap();
                bootstrap.group(bossGroup, workerGroup)
                        .channel(NioServerSocketChannel.class)
                        .option(ChannelOption.SO_BACKLOG, 1024)
                        .childHandler(new ChannelInitializer<SocketChannel>() { 
                            @Override
                            protected void initChannel(SocketChannel socketChannel) throws Exception { 
                             socketChannel.pipeline().addLast(new MyNettyServerHandler());
                            }
                        });

方法 `.group(bossGroup, workerGroup)` 这里 Netty 称之为 parentGroup 和 childGroup ，即 bossGroup = parentGroup 、workerGroup = childGroup 。赋值时 childGroup 直接对变量 `this.childGroup` 赋值，parentGroup 调用了父类的构造方法 `super.group(parentGroup);`

/** * Set the {@link EventLoopGroup} for the parent (acceptor) and the child (client). These * {@link EventLoopGroup}'s are used to handle all the events and IO for {@link ServerChannel} and * {@link Channel}'s. */
        public ServerBootstrap group(EventLoopGroup parentGroup, EventLoopGroup childGroup) { 
            super.group(parentGroup);
            if (childGroup == null) { 
                throw new NullPointerException("childGroup");
            }
            if (this.childGroup != null) { 
                throw new IllegalStateException("childGroup set already");
            }
            this.childGroup = childGroup;
            return this;
        }

代码 `super.group(parentGroup);` 即给父类 `AbstractBootstrap` 变量 `volatile EventLoopGroup group;` 进行了初始化赋值，注释也写的很明确了：

> The \{@link EventLoopGroup\} which is used to handle all the events for the to-be-created  
> 这个 EventLoopGroup 用于处理所有要被创建的事件

public abstract class AbstractBootstrap
    
    	volatile EventLoopGroup group;
        /** * The {@link EventLoopGroup} which is used to handle all the events for the to-be-created * {@link Channel} */
        public B group(EventLoopGroup group) { 
            if (group == null) { 
                throw new NullPointerException("group");
            }
            if (this.group != null) { 
                throw new IllegalStateException("group set already");
            }
            this.group = group;
            return self();
        }

这里我们逻辑上的 bossGroup 和 workGroup 对应到 ServerBootstrap 是成员 group 和 childGroup

方法 `.channel(NioServerSocketChannel.class)`

/** * The {@link Class} which is used to create {@link Channel} instances from. * You either use this or {@link #channelFactory(io.netty.channel.ChannelFactory)} if your * {@link Channel} implementation has no no-args constructor. */
        public B channel(Class<? extends C> channelClass) { 
            if (channelClass == null) { 
                throw new NullPointerException("channelClass");
            }
            return channelFactory(new ReflectiveChannelFactory<C>(channelClass));
        }

这个方法使其就是在给 `class ServerBootstrap extends AbstractBootstrap<ServerBootstrap, ServerChannel>` 的变量 `private volatile ChannelFactory<? extends C> channelFactory;` 赋值  
他实例化了一个我们传入类的反射工厂 `new ReflectiveChannelFactory<C>(channelClass)` 这个反射工厂其实很简单，我们稍微看下它核心的两个方法其实也就这两个方法和一个 toString()

public ReflectiveChannelFactory(Class<? extends T> clazz) { 
            ObjectUtil.checkNotNull(clazz, "clazz");
            try { 
                this.constructor = clazz.getConstructor();
            } catch (NoSuchMethodException e) { 
                throw new IllegalArgumentException("Class " + StringUtil.simpleClassName(clazz) +
                        " does not have a public non-arg constructor", e);
            }
        }
    
        @Override
        public T newChannel() { 
            try { 
                return constructor.newInstance();
            } catch (Throwable t) { 
                throw new ChannelException("Unable to create Channel from class " + constructor.getDeclaringClass(), t);
            }
        }

一个赋值 构造方法 `this.constructor = clazz.getConstructor();` 一个 new Channel 实例 `return constructor.newInstance();` 我们这里初始化了构造器，那么在后面的某些时候一定会进行它的实例化，即调用 `this.channelFactory.newChannel()` 这里我们要大概有个印象。

到了这里我相信大家都有些感觉了，这一套链式编程就是在给成员变量赋值而已，不是在给 ServerBootstrap 的成员变量赋值，就是在给其抽象父类 AbstractBootstrap 的成员变量赋值仅此而已，没啥难的，应该很好懂了。

来看 `.option(ChannelOption.SO_BACKLOG, 1024)` 这个其实就是在给 键值对 `Map<ChannelOption<?>, Object> options` 赋值，给我们的 ServerBootstrap 进行配置，预留的配置都在 `ChannelOption` 类中给出了。

public abstract class AbstractBootstrap
    
    private final Map<ChannelOption<?>, Object> options = new LinkedHashMap<ChannelOption<?>, Object>();
    
    
        /** * Allow to specify a {@link ChannelOption} which is used for the {@link Channel} instances once they got * created. Use a value of {@code null} to remove a previous set {@link ChannelOption}. */
        public <T> B option(ChannelOption<T> option, T value) { 
            if (option == null) { 
                throw new NullPointerException("option");
            }
            if (value == null) { 
                synchronized (options) { 
                    options.remove(option);
                }
            } else { 
                synchronized (options) { 
                    options.put(option, value);
                }
            }
            return self();
        }

最后一块，初始化我们定制的 channelPipeline 链路基本上也是固定的写法了。最后给变量 `childHandler` 进行了赋值

/** * Set the {@link ChannelHandler} which is used to serve the request for the {@link Channel}'s. */
        public ServerBootstrap childHandler(ChannelHandler childHandler) { 
            if (childHandler == null) { 
                throw new NullPointerException("childHandler");
            }
            this.childHandler = childHandler;
            return this;
        }

**最后总结一下进行的几个关键变量的赋值：**

> EventLoopGroup ServerBootstrap.childGroup (workerGroup 事件接收处理)  
> EventLoopGroup AbstractBootstrap.group （bossGroup 事件创建）  
> ChannelFactory<? extends C> AbstractBootstrap.channelFactory （channel 反射工厂，最后用于实例化 C）  
> Map<ChannelOption<?>, Object> AbstractBootstrap.options （Bootstrap 的配置项）  
> ChannelHandler ServerBootstrap.childHandler （通道处理器，初始化我们自定义的处理器链路）

### 最后 ###

本篇带你过了 Netty 服务端源码的前两部分 NioEventLoopGroup 的初始化过程，以及第二部分的 ServerBootstrap 的关键属性赋值，同时介绍了其作用。在第三篇中我将重点带大家梳理最关键的步骤 `ChannelFuture cf = bootstrap.bind(9000).sync();` bind 绑定端口过程。最后希望本文对你有所帮助，我们 Netty （三）不见不散。

**我是 [dying 搁浅][dying] ，我始终期待与你的相遇。无论你是否期待，潮涨潮落，我仅且就在这里…………**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3c5MDMzMjg2MTU_size_16_color_FFFFFF_t_70 6]

[Netty _ Netty]: https://blog.csdn.net/w903328615/article/details/114899473
[Netty _ Netty _ new NioEventLoopGroup_nThreads_]: https://blog.csdn.net/w903328615/article/details/115680055
[Netty _ Netty _initAndRegister_ _ init_]: https://blog.csdn.net/w903328615/article/details/115982104
[Netty _ Netty _initAndRegister_ _ init_ 1]: https://blog.csdn.net/w903328615/article/details/116499703
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3c5MDMzMjg2MTU_size_16_color_FFFFFF_t_70]: /images/20221022/9aee29a95439431b9375bb4b90b30797.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3c5MDMzMjg2MTU_size_16_color_FFFFFF_t_70 1]: /images/20221022/4cfdf9b55f75486f99724ecf7d37ab11.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3c5MDMzMjg2MTU_size_16_color_FFFFFF_t_70 2]: /images/20221022/13ed23f8edd94a89a13246d0eaa6905c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3c5MDMzMjg2MTU_size_16_color_FFFFFF_t_70 3]: /images/20221022/d73c9c43f43b424c8d3dcbcb5bf1c737.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3c5MDMzMjg2MTU_size_16_color_FFFFFF_t_70 4]: /images/20221022/0fef93cab4284416b86620960187e57e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3c5MDMzMjg2MTU_size_16_color_FFFFFF_t_70 5]: /images/20221022/a0d1e0a1377c4f16b9474fc4aaec66b8.png
[dying]: https://blog.csdn.net/w903328615
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3c5MDMzMjg2MTU_size_16_color_FFFFFF_t_70 6]: /images/20221022/ee325a0b83724f2fb94717ae67ffc6fe.png