摘要：關于的線程模型首先我們來看一下的線程模型的線程模型有三種單線程模型多線程模型主從多線程模型首先來看一下單線程模型所謂單線程即處理和處理都在一個線程中處理這個模型的壞處顯而易見當其中某個阻塞時會導致其他所有的的都得不到執(zhí)行并且更嚴重的是的阻塞

首先我們來看一下 Reactor 的線程模型.
Reactor 的線程模型有三種:

單線程模型

多線程模型

主從多線程模型

首先來看一下 單線程模型:

所謂單線程, 即 acceptor 處理和 handler 處理都在一個線程中處理. 這個模型的壞處顯而易見: 當其中某個 handler 阻塞時, 會導致其他所有的 client 的 handler 都得不到執(zhí)行, 并且更嚴重的是, handler 的阻塞也會導致整個服務不能接收新的 client 請求(因為 acceptor 也被阻塞了). 因為有這么多的缺陷, 因此單線程 Reactor 模型用的比較少.

那么什么是多線程模型呢? Reactor 的多線程模型與單線程模型的區(qū)別就是 acceptor 是一個多帶帶的線程處理, 并且有一組特定的 NIO 線程來負責各個客戶端連接的 IO 操作. Reactor 多線程模型如下:

Reactor 多線程模型 有如下特點:

有專門一個線程, 即 Acceptor 線程用于監(jiān)聽客戶端的TCP連接請求.

客戶端連接的 IO 操作都是由一個特定的 NIO 線程池負責. 每個客戶端連接都與一個特定的 NIO 線程綁定, 因此在這個客戶端連接中的所有 IO 操作都是在同一個線程中完成的.

客戶端連接有很多, 但是 NIO 線程數(shù)是比較少的, 因此一個 NIO 線程可以同時綁定到多個客戶端連接中.

接下來我們再來看一下 Reactor 的主從多線程模型.

一般情況下, Reactor 的多線程模式已經(jīng)可以很好的工作了, 但是我們考慮一下如下情況: 如果我們的服務器需要同時處理大量的客戶端連接請求或我們需要在客戶端連接時, 進行一些權限的檢查, 那么單線程的 Acceptor 很有可能就處理不過來, 造成了大量的客戶端不能連接到服務器.

Reactor 的主從多線程模型就是在這樣的情況下提出來的, 它的特點是: 服務器端接收客戶端的連接請求不再是一個線程, 而是由一個獨立的線程池組成. 它的線程模型如下:

可以看到, Reactor 的主從多線程模型和 Reactor 多線程模型很類似, 只不過 Reactor 的主從多線程模型的 acceptor 使用了線程池來處理大量的客戶端請求.

我們介紹了三種 Reactor 的線程模型, 那么它們和 NioEventLoopGroup 又有什么關系呢? 其實, 不同的設置 NioEventLoopGroup 的方式就對應了不同的 Reactor 的線程模型.

來看一下下面的例子:

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

注意, 我們實例化了一個 NioEventLoopGroup, 構造器參數(shù)是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 和后續(xù)的所有客戶端連接的 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 核心數(shù)乘以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 是用于服務端的 accept, 即用于處理客戶端的連接請求. 我們可以把 Netty 比作一個飯店, bossGroup 就像一個像一個前臺接待, 當客戶來到飯店吃時, 接待員就會引導顧客就坐, 為顧客端茶送水等.

而 workerGroup, 其實就是實際上干活的, 它們負責客戶端連接通道的 IO 操作: 當接待員 招待好顧客后, 就可以稍做休息, 而此時后廚里的廚師們(workerGroup)就開始忙碌地準備飯菜了.

關于 bossGroup 與 workerGroup 的關系, 我們可以用如下圖來展示:

首先, 服務器端 bossGroup 不斷地監(jiān)聽是否有客戶端的連接, 當發(fā)現(xiàn)有一個新的客戶端連接到來時, bossGroup 就會為此連接初始化各項資源, 然后從 workerGroup 中選出一個 EventLoop 綁定到此客戶端連接中. 那么接下來的服務器與客戶端的交互過程就全部在此分配的 EventLoop 中了.

NioEventLoop 繼承于 SingleThreadEventLoop, 而 SingleThreadEventLoop 又繼承于 SingleThreadEventExecutor. SingleThreadEventExecutor 是 Netty 中對本地線程的抽象, 它內部有一個 Thread thread 屬性, 存儲了一個本地 Java 線程. 因此我們可以認為, 一個 NioEventLoop 其實和一個特定的線程綁定, 并且在其生命周期內, 綁定的線程都不會再改變.

NioEventLoop -> SingleThreadEventLoop -> SingleThreadEventExecutor -> AbstractScheduledEventExecutor

在 AbstractScheduledEventExecutor 中, Netty 實現(xiàn)了 NioEventLoop 的 schedule 功能, 即我們可以通過調用一個 NioEventLoop 實例的 schedule 方法來運行一些定時任務. 而在 SingleThreadEventLoop 中, 又實現(xiàn)了任務隊列的功能, 通過它, 我們可以調用一個 NioEventLoop 實例的 execute 方法來向任務隊列中添加一個 task, 并由 NioEventLoop 進行調度執(zhí)行.

通常來說, NioEventLoop 肩負著兩種任務, 第一個是作為 IO 線程, 執(zhí)行與 Channel 相關的 IO 操作, 包括 調用 select 等待就緒的 IO 事件、讀寫數(shù)據(jù)與數(shù)據(jù)的處理等; 而第二個任務是作為任務隊列, 執(zhí)行 taskQueue 中的任務, 例如用戶調用 eventLoop.schedule 提交的定時任務也是這個線程執(zhí)行的.

EventLoopGroup(其實是MultithreadEventExecutorGroup) 內部維護一個類型為 EventExecutor children 數(shù)組, 其大小是 nThreads, 這樣就構成了一個線程池

如果我們在實例化 NioEventLoopGroup 時, 如果指定線程池大小, 則 nThreads 就是指定的值, 反之是處理器核心數(shù) * 2

MultithreadEventExecutorGroup 中會調用 newChild 抽象方法來初始化 children 數(shù)組

抽象方法 newChild 是在 NioEventLoopGroup 中實現(xiàn)的, 它返回一個 NioEventLoop 實例

NioEventLoopGroup 就像一個線程池, 負責為每個新創(chuàng)建的 Channel 分配一個 EventLoop. 而 EventLoop 就是一個線程, 負責執(zhí)行用戶任務和 IO 事件.
值得注意的是: 執(zhí)行 IO 事件(自己的業(yè)務邏輯)時, 如果當前業(yè)務邏輯沒有執(zhí)行完畢, 是無法處理下一個 IO 事件的.

我們已經(jīng)提到過, 在Netty 中, 一個 NioEventLoop 通常需要肩負起兩種任務, 第一個是作為 IO 線程, 處理 IO 操作; 第二個就是作為任務線程, 處理 taskQueue 中的任務.

NioEventLoop 繼承于 SingleThreadEventExecutor, 而 SingleThreadEventExecutor 中有一個 Queue taskQueue 字段, 用于存放添加的 Task. 在 Netty 中, 每個 Task 都使用一個實現(xiàn)了 Runnable 接口的實例來表示.
例如當我們需要將一個 Runnable 添加到 taskQueue 中時, 我們可以進行如下操作:

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

當一個任務被添加到 taskQueue 后, 它是怎么被 EventLoop 執(zhí)行的呢?
讓我們回到 NioEventLoop.run() 方法中, 在這個方法里, 會分別調用 processSelectedKeys() 和 runAllTasks() 方法, 來進行 IO 事件的處理和 task 的處理.
runAllTasks 方法有兩個重載的方法, 一個是無參數(shù)的, 另一個有一個參數(shù)的. 首先來看一下無參數(shù)的 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 中已經(jīng)可以執(zhí)行的(即定時時間已到的 schedule 任務) 拿出來并添加到 taskQueue 中, 作為可執(zhí)行的 task 等待被調度執(zhí)行.

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 中獲取一個可執(zhí)行的 task, 然后調用它的 run() 方法來運行此 task.