EventLoop 和 线程模型

关于 Reactor 的线程模型

首先咱们来看一下 Reactor 的线程模型.
Reactor 的线程模型有三种:

• 单线程模型
• 多线程模型
• 主从多线程模型

首先来看一下 单线程模型:

所谓单线程, 即 acceptor 处理和 handler 处理都在一个线程中处理. 这个模型的坏处显而易见: 当其中某个 handler 阻塞时, 会致使其余全部的 client 的 handler 都得不到执行, 而且更严重的是, handler 的阻塞也会致使整个服务不能接收新的 client 请求(由于 acceptor 也被阻塞了). 由于有这么多的缺陷, 所以单线程 Reactor 模型用的比较少.

那么什么是多线程模型呢? Reactor 的多线程模型与单线程模型的区别就是 acceptor 是一个单独的线程处理, 而且有一组特定的 NIO 线程来负责各个客户端链接的 IO 操做. Reactor 多线程模型以下:

Reactor 多线程模型 有以下特色:

• 有专门一个线程, 即 Acceptor 线程用于监听客户端的TCP链接请求.
• 客户端链接的 IO 操做都是由一个特定的 NIO 线程池负责. 每一个客户端链接都与一个特定的 NIO 线程绑定, 所以在这个客户端链接中的全部 IO 操做都是在同一个线程中完成的.
• 客户端链接有不少, 可是 NIO 线程数是比较少的, 所以一个 NIO 线程能够同时绑定到多个客户端链接中.

接下来咱们再来看一下 Reactor 的主从多线程模型.

通常状况下, Reactor 的多线程模式已经能够很好的工做了, 可是咱们考虑一下以下状况: 若是咱们的服务器须要同时处理大量的客户端链接请求或咱们须要在客户端链接时, 进行一些权限的检查, 那么单线程的 Acceptor 颇有可能就处理不过来, 形成了大量的客户端不能链接到服务器.

Reactor 的主从多线程模型就是在这样的状况下提出来的, 它的特色是: 服务器端接收客户端的链接请求再也不是一个线程, 而是由一个独立的线程池组成. 它的线程模型以下:

能够看到, Reactor 的主从多线程模型和 Reactor 多线程模型很相似, 只不过 Reactor 的主从多线程模型的 acceptor 使用了线程池来处理大量的客户端请求.

NioEventLoopGroup 与 Reactor 线程模型的对应

咱们介绍了三种 Reactor 的线程模型, 那么它们和 NioEventLoopGroup 又有什么关系呢? 其实, 不一样的设置 NioEventLoopGroup 的方式就对应了不一样的 Reactor 的线程模型.

单线程模型

来看一下下面的例子:

EventLoopGroup bossGroup = new NioEventLoopGroup(1);
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup)
 .channel(NioServerSocketChannel.class)
 ...

注意, 咱们实例化了一个 NioEventLoopGroup, 构造器参数是1, 表示 NioEventLoopGroup 的线程池大小是1.

而后接着咱们调用 b.group(bossGroup) 设置了服务器端的 EventLoopGroup. 有些朋友可能会有疑惑: 我记得在启动服务器端的 Netty 程序时, 是须要设置 bossGroup 和 workerGroup 的, 为何这里就只有一个 bossGroup?

其实很简单, ServerBootstrap 重写了 group 方法:

@Override
public ServerBootstrap group(EventLoopGroup group) {
    return group(group, group);
}

所以当传入一个 group 时, 那么 bossGroup 和 workerGroup 就是同一个 NioEventLoopGroup 了. 而且这个 NioEventLoopGroup 只有一个线程, 这样就会致使 Netty 中的 acceptor 和后续的全部客户端链接的 IO 操做都是在一个线程中处理的. 那么对应到 Reactor 的线程模型中, 咱们这样设置 NioEventLoopGroup 时, 就至关于 Reactor 单线程模型.

多线程模型

同理, 再来看一下下面的例子:

EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
 .channel(NioServerSocketChannel.class)
 ...

bossGroup 中只有一个线程, 而 workerGroup 中的线程是 CPU 核心数乘以2, 所以对应的到 Reactor 线程模型中, 咱们知道, 这样设置的 NioEventLoopGroup 其实就是 Reactor 多线程模型.

主从多线程模型

EventLoopGroup bossGroup = new NioEventLoopGroup(4);
EventLoopGroup workerGroup = new NioEventLoopGroup();
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
 .channel(NioServerSocketChannel.class)
 ...

服务器端的 ServerSocketChannel 只绑定到了 bossGroup 中的一个线程, 所以在调用 Java NIO 的 Selector.select 处理客户端的链接请求时, 其实是在一个线程中的, 因此对只有一个服务的应用来讲, bossGroup 设置多个线程是没有什么做用的, 反而还会形成资源浪费.

关于 bossGroup 与 workerGroup

bossGroup 是用于服务端的 accept, 即用于处理客户端的链接请求. 咱们能够把 Netty 比做一个饭店, bossGroup 就像一个像一个前台接待, 当客户来到饭店吃时, 接待员就会引导顾客就坐, 为顾客端茶送水等.

而 workerGroup, 其实就是实际上干活的, 它们负责客户端链接通道的 IO 操做: 当接待员 招待好顾客后, 就能够稍作休息, 而此时后厨里的厨师们(workerGroup)就开始忙碌地准备饭菜了.

关于 bossGroup 与 workerGroup 的关系, 咱们能够用以下图来展现:

首先, 服务器端 bossGroup 不断地监听是否有客户端的链接, 当发现有一个新的客户端链接到来时, bossGroup 就会为此链接初始化各项资源, 而后从 workerGroup 中选出一个 EventLoop 绑定到此客户端链接中. 那么接下来的服务器与客户端的交互过程就所有在此分配的 EventLoop 中了.

NioEventLoop

NioEventLoop 继承于 SingleThreadEventLoop, 而 SingleThreadEventLoop 又继承于 SingleThreadEventExecutor. SingleThreadEventExecutor 是 Netty 中对本地线程的抽象, 它内部有一个 Thread thread 属性, 存储了一个本地 Java 线程. 所以咱们能够认为, 一个 NioEventLoop 其实和一个特定的线程绑定, 而且在其生命周期内, 绑定的线程都不会再改变.

NioEventLoop -> SingleThreadEventLoop -> SingleThreadEventExecutor -> AbstractScheduledEventExecutor

在 AbstractScheduledEventExecutor 中, Netty 实现了 NioEventLoop 的 schedule 功能, 即咱们能够经过调用一个 NioEventLoop 实例的 schedule 方法来运行一些定时任务. 而在 SingleThreadEventLoop 中, 又实现了任务队列的功能, 经过它, 咱们能够调用一个 NioEventLoop 实例的 execute 方法来向任务队列中添加一个 task, 并由 NioEventLoop 进行调度执行.

一般来讲, NioEventLoop 肩负着两种任务, 第一个是做为 IO 线程, 执行与 Channel 相关的 IO 操做, 包括 调用 select 等待就绪的 IO 事件、读写数据与数据的处理等; 而第二个任务是做为任务队列, 执行 taskQueue 中的任务, 例如用户调用 eventLoop.schedule 提交的定时任务也是这个线程执行的.

NioEventLoopGroup

• EventLoopGroup(实际上是MultithreadEventExecutorGroup) 内部维护一个类型为 EventExecutor children 数组, 其大小是 nThreads, 这样就构成了一个线程池
• 若是咱们在实例化 NioEventLoopGroup 时, 若是指定线程池大小, 则 nThreads 就是指定的值, 反之是处理器核心数 * 2
• MultithreadEventExecutorGroup 中会调用 newChild 抽象方法来初始化 children 数组
• 抽象方法 newChild 是在 NioEventLoopGroup 中实现的, 它返回一个 NioEventLoop 实例

NioEventLoopGroup 就像一个线程池, 负责为每一个新建立的 Channel 分配一个 EventLoop. 而 EventLoop 就是一个线程, 负责执行用户任务和 IO 事件.

Netty 的任务队列机制

咱们已经提到过, 在Netty 中, 一个 NioEventLoop 一般须要肩负起两种任务, 第一个是做为 IO 线程, 处理 IO 操做; 第二个就是做为任务线程, 处理 taskQueue 中的任务.

Task 的添加

普通 Runnable 任务

NioEventLoop 继承于 SingleThreadEventExecutor, 而 SingleThreadEventExecutor 中有一个 Queue<Runnable> taskQueue 字段, 用于存放添加的 Task. 在 Netty 中, 每一个 Task 都使用一个实现了 Runnable 接口的实例来表示.
例如当咱们须要将一个 Runnable 添加到 taskQueue 中时, 咱们能够进行以下操做:

EventLoop eventLoop = channel.eventLoop();
eventLoop.execute(new Runnable() {
    @Override
    public void run() {
        System.out.println("Hello, Netty!");
    }
});

任务的执行

当一个任务被添加到 taskQueue 后, 它是怎么被 EventLoop 执行的呢?
让咱们回到 NioEventLoop.run() 方法中, 在这个方法里, 会分别调用 processSelectedKeys() 和 runAllTasks() 方法, 来进行 IO 事件的处理和 task 的处理.
runAllTasks 方法有两个重载的方法, 一个是无参数的, 另外一个有一个参数的. 首先来看一下无参数的 runAllTasks:

protected boolean runAllTasks() {
    fetchFromScheduledTaskQueue();
    Runnable task = pollTask();
    if (task == null) {
        return false;
    }

    for (;;) {
        try {
            task.run();
        } catch (Throwable t) {
            logger.warn("A task raised an exception.", t);
        }

        task = pollTask();
        if (task == null) {
            lastExecutionTime = ScheduledFutureTask.nanoTime();
            return true;
        }
    }
}

在此方法的一开始调用的 fetchFromScheduledTaskQueue() 其实就是将 scheduledTaskQueue 中已经能够执行的(即定时时间已到的 schedule 任务) 拿出来并添加到 taskQueue 中, 做为可执行的 task 等待被调度执行.

private void fetchFromScheduledTaskQueue() {
    if (hasScheduledTasks()) {
        long nanoTime = AbstractScheduledEventExecutor.nanoTime();
        for (;;) {
            Runnable scheduledTask = pollScheduledTask(nanoTime);
            if (scheduledTask == null) {
                break;
            }
            taskQueue.add(scheduledTask);
        }
    }
}

接下来 runAllTasks() 方法就会不断调用 task = pollTask() 从 taskQueue 中获取一个可执行的 task, 而后调用它的 run() 方法来运行此 task.