Netty的线程模型

java

Publish Date: 2021-01-10

Netty是一个异步、基于事件驱动的网络应用程序框架，其对 Java NIO进行了封装，大大简化了 TCP 或者 UDP 服务器的网络编程。它的设计参考了许多协议的实现，比如 FTP，SMTP，HTTP 和各种二进制和基于文本的传统协议，因此 Netty 成功的实现了兼顾快速开发，性能，稳定性，灵活性为一体，不需要为了考虑一方面原因而妥协其他方面，因此应用比较广泛。一般首先会想到的是经典的Reactor线程模型，尽管不同的NIO框架对于Reactor模式的实现存在差异，但本质上还是遵循了Reactor的基础线程模型。

Netty是Java领域有名的开源网络库，很多流行的框架都是基于它来构建的，比如Apache Dubbo 、Apache RocketMq、Zuul 2.0服务网关、Spring WebFlux、Sofa-Bolt 、Hadoop底层网络通讯都是基于 Netty 来实现的。当我们讨论Netty线程模型的时候

Java IO模型

Netty是基于Java的NIO实现，对于IO模型的实现原理涉及到内核且不是本文重点，不详说了，主要简介一下几种IO模型

BIO：同步阻塞IO模型；
NIO：基于IO多路复用技术的“非阻塞同步”IO模型。简单来说，内核将可读可写事件通知应用，由应用主动发起读写操作；
AIO：非阻塞异步IO模型。简单来说，内核将读完成事件通知应用，读操作由内核完成，应用只需操作数据即可；应用做异步写操作时立即返回，内核会进行写操作排队并执行写操作。

NIO和AIO不同之处在于应用线程是否进行真正的读写操作。NIO都需要应用线程主动去内核空间把数据读到用户空间，而AIO基于则不用

在说Reactor线程模型之前，我们需要先对基本并发编程模型：串行工作者模型、并发工作者模型进行讲解，因为netty中的reactor线程模型是在这个基础之上演进的，并且Netty 的线程模型并不是一成不变的，它实际取决于用户的启动参数配置。通过设置不同的启动参数，Netty 可以同时支持 Reactor 单线程模型、多线程模型。

基本并发编程模型

串行工作者模型和并行工作者模型关注的是将任务划分为2个阶段：1、任务的接受阶段 2、任务的处理阶段；而reactor线程模型关注的是上述第二个阶段：任务在处理的过程中，继续划分为多个步骤进行处理

串行工作者模型

我们以一个典型的任务处理流程，来说明为什么要将任务的接受流程与处理流程划分开来

这样的方式我们可以用Java的单个线程的线程池实现，这样做存在的问题是，必须要等待某个task处理完成之后，才能接受处理下一个task。一般而言，任务的处理过程会比任务的接受流程慢得多。例如在处理任务的时候，我们可能会需要访问远程数据库，这属于一种网络IO。通常情况下IO操作是比较耗时的，这直接影响了下一个任务的接受，而且通常在IO操作的时候,CPU是比较空闲的，白白浪费了资源。对于优化方式，可能你首先想到的就增加线程池的线程个数，通过多线程的方式提供性能，不过这样的话就已经变成了并发工作者模型，对于上例的串行任务还有优化的余地，就是考虑将任务的接受与处理分为两个线程进行处理，一个只负责接受任务，一个只负责处理任务。这就演化出了第一个线程模型：串行工作者模型

接受任务的线程称之为AcceptThread，其将接受到的任务放到一个任务队列中，因此能立即返回接受下一个任务。而worker线程不断的从这个队列中取出任务进行异步执行。目前这种情况存在一个很大的问题，在于任务处理的太慢，导致队列里积压的任务数量越来愈大，任务不能得到及时的执行。所以我们可以用多个worker thread来处理任务。

并行工作者模型

在并行工作者模型中，有一个accpet thread，多个worker thread，因为worker thread的功能都相同，所以我们通常会将其划分到一个组中(worker thread group)或者说资源池中。并行工作者线程模型有两种设计方式，以下分别进行介绍。

基于公共任务队列

accept thread将接受到的任务放到任务队列中，worker thread group中的多个 worker thread 并行的从公共的队列中拉取任务进行处理

右边的Work Thread Group可以用Java的线程池实现，通过在main线程中接受任务，将任务提交到线程池中，即可以完成上述线程模型。

基于私有任务队列

在基于公共任务队列中，由于多个worker线程同时从一个公共的任务队列中拉取任务进行处理，因此必须要进行加锁，因而影响了效率。因此又有了下面一种设计方式：reactor thread直接将任务转发给各个worker thread，每个worker thread内部维护一个队列来处理

这种方式的设计，由于使用了私有的任务队列，避免的锁竞争进一步提高了性能，因为每个worker thread都从各自的队列中取出任务进行执行。实际上，在netty的实现中，就是为每个worker thread维护了一个队列。需要注意的是：在基于公共任务队列的实现中accpet thread直接接收任务然后再把任务丢到公共队列就可以了。而在基于私有任务队列的实现中，accpet thread不仅仅需要接收任务，而需要把任务平均分配到各个work线程的私有队列中。

Reactor线程模型

reactor线程模型并不是netty所独有，其是一种并发编程模型，更确切的或者说一种思想，其具有的是指导意义，开发者需要在这种编程模型思想的指导下，结合自己的实际场景，来进行合理的设计。在不同的场景下，可能设计出来的reactor线程模型是不一样的，例如scala中的akka框架，就是基于reactor线程模型的思想设计的。netty只是结合了nio网络编程的特点，合理的应用了reactor线程模型。

串行工作者模型和并行工作者模型，它们主要的关注点是：划分任务的接受阶段与任务的处理阶段。也正是因为如此，我们通常将接受任务的线程称之为Accpet Thread。而任务的处理过程都是一个线程(worker thread)内完成的。reactor线程模型关注的是：任务接受之后，对处理过程继续进行切分，划分为多个不同的步骤，每个步骤用不同的线程来处理，也就是原本由一个线程处理的任务现在由多个线程来处理，每个线程在处理完自己的步骤之后，还需要将任务转发到线程继续进行处理。为了进行区分，在reactor线程模型中，处理任务并且分发的线程，不再称之为worker thread，而是reactor thread。

单线程reactor线程模型

下图演示了单线程reactor线程模型，之所以称之为单线程，还是因为只有一个accpet Thread接受任务，之后转发到reactor线程中进行处理。右边两个框表示的是Reactor Thread Group，里面有多个Reactor Thread。一个Reactor Thread Group中的Reactor Thread功能都是相同的，例如第一个黄色框中的Reactor Thread都是处理拆分后的任务的第一阶段，第二个框中的Reactor Thread都是处理拆分后的任务的第二步骤。任务具体要怎么拆分，要结合具体场景，下图只是演示作用。一般来说，都是以比较耗时的操作(例如IO)为切分点。

如果我们在任务处理的过程中，不划分为多个阶段进行处理的话，那么单线程reactor线程模型就退化成了并行工作和模型。事实上，可以认为并行工作者模型，就是单线程reactor线程模型的最简化版本。

多线程reactor线程模型

在前面所有的模型中，accpet线程模型始终只有一个，工作线程都做了池化异步处理，在高并发场景下单线程的accpet就成了性能瓶颈所在，所谓多线程reactor线程模型，无非就是有多个accpet线程，解决accpet的压力。

reactor线程模型的本质：

将任务处理切分成多个阶段进行，每个阶段处理完自己的部分之后，转发到下一个阶段进行处理。不同的阶段之间的执行是异步的，可以认为每个阶段都有一个独立的线程池。
不同的类型的任务，有着不同的处理流程，划分时需要根据场景划分成不同的阶段

Netty中的Reactor线程模型

在我们编写的基于netty的应用中都有可能出现，甚至可能会不用reactor线程。具体属于哪一种情况，要看我们的代码是如何编写的。首先看一下Netty服务端启动的代码：

         // Configure the server.
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
             .channel(NioServerSocketChannel.class)
             .option(ChannelOption.SO_BACKLOG, 100)
             .handler(new LoggingHandler(LogLevel.INFO))
             .childHandler(new ChannelInitializer<SocketChannel>() {
                 @Override
                 public void initChannel(SocketChannel ch) throws Exception {
                     ChannelPipeline p = ch.pipeline();
                     p.addLast(your_handler_name, your_handler_instance);
                 }
             })
             .childOption(ChannelOption.SO_KEEPALIVE, true); 
             .childAttr(AttributeKey.valueOf("sc.key"),"sc.value");

            // Start the server.
            ChannelFuture f = b.bind(PORT).sync();

            // Wait until the server socket is closed.
            f.channel().closeFuture().sync();
        } finally {
            // Shut down all event loops to terminate all threads.
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }

在上述代码片段中代码很少，却包含了一个复杂reactor线程模型，其工作流程如下图

这里的代码是不是很像函数式编程的风格，方法调完之后可以接着点它的其他的方法，这里实际使用了 Builder 设计模式

Netty抽象出两组线程池 BossGroup 专门负责接收客户端的连接, WorkerGroup 专门负责网络的读写
BossGroup 和 WorkerGroup 类型都是 NioEventLoopGroup
NioEventLoopGroup 相当于一个事件循环组, 这个组中含有多个事件循环，每一个事件循环是 NioEventLoop
NioEventLoop 表示一个不断循环的执行处理任务的线程，每个NioEventLoop 都有一个selector , 用于监听绑定在其上的socket的网络通讯
NioEventLoopGroup 可以有多个线程, 即可以含有多个NioEventLoop
每个Boss中NioEventLoop 循环执行的步骤有3步
1. 轮询accept 事件
2. 处理accept 事件 , 与client建立连接 , 生成NioScocketChannel , 并将其注册到某个worker中NIOEventLoop上的selector
3. 处理任务队列的任务 ,即 runAllTasks
每个 Worker中NIOEventLoop 循环执行的步骤
1. 轮询read, write 事件
2. 处理i/o事件，即read , write 事件，在对应NioScocketChannel 处理
3. 处理任务队列的任务，即 runAllTasks
每个Worker NIOEventLoop 处理业务时，会使用pipeline(管道), pipeline 中包含了 channel , 即通过pipeline 可以获取到对应通道, 管道中维护了很多的处理器

Netty各模块之间的关系

Netty 抽象出两组线程池，BossGroup 专门负责接收客户端连接，WorkerGroup 专门负责网络读写操作；
NioEventLoop 表示一个不断循环执行处理任务的线程，每个 NioEventLoop 都有一个 selector，用于监听绑定在其上的 socket 网络通道；
NioEventLoop 内部采用串行化设计，从消息的读取->解码->处理->编码->发送，始终由 IO 线程 NioEventLoop 负责；
NioEventLoopGroup 下包含多个 NioEventLoop；
每个 NioEventLoop 中包含有一个 Selector，一个 taskQueue；
每个 NioEventLoop 的 Selector 上可以注册监听多个 NioChannel；
每个 NioChannel 只会绑定在唯一的 NioEventLoop 上；
每个 NioChannel 都绑定有一个自己的 ChannelPipeline。