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

首先我們來(lái)看一下 Reactor 的線程模型.
Reactor 的線程模型有三種:

單線程模型

多線程模型

主從多線程模型

首先來(lái)看一下 單線程模型:

所謂單線程, 即 acceptor 處理和 handler 處理都在一個(gè)線程中處理. 這個(gè)模型的壞處顯而易見: 當(dāng)其中某個(gè) handler 阻塞時(shí), 會(huì)導(dǎo)致其他所有的 client 的 handler 都得不到執(zhí)行, 并且更嚴(yán)重的是, handler 的阻塞也會(huì)導(dǎo)致整個(gè)服務(wù)不能接收新的 client 請(qǐng)求(因?yàn)?acceptor 也被阻塞了). 因?yàn)橛羞@么多的缺陷, 因此單線程 Reactor 模型用的比較少.

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

Reactor 多線程模型 有如下特點(diǎn):

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

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

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

接下來(lái)我們?cè)賮?lái)看一下 Reactor 的主從多線程模型.

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

Reactor 的主從多線程模型就是在這樣的情況下提出來(lái)的, 它的特點(diǎn)是: 服務(wù)器端接收客戶端的連接請(qǐng)求不再是一個(gè)線程, 而是由一個(gè)獨(dú)立的線程池組成. 它的線程模型如下:

可以看到, Reactor 的主從多線程模型和 Reactor 多線程模型很類似, 只不過(guò) Reactor 的主從多線程模型的 acceptor 使用了線程池來(lái)處理大量的客戶端請(qǐng)求.

我們介紹了三種 Reactor 的線程模型, 那么它們和 NioEventLoopGroup 又有什么關(guān)系呢? 其實(shí), 不同的設(shè)置 NioEventLoopGroup 的方式就對(duì)應(yīng)了不同的 Reactor 的線程模型.

來(lái)看一下下面的例子:

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

注意, 我們實(shí)例化了一個(gè) NioEventLoopGroup, 構(gòu)造器參數(shù)是1, 表示 NioEventLoopGroup 的線程池大小是1.

然后接著我們調(diào)用 b.group(bossGroup) 設(shè)置了服務(wù)器端的 EventLoopGroup. 有些朋友可能會(huì)有疑惑: 我記得在啟動(dòng)服務(wù)器端的 Netty 程序時(shí), 是需要設(shè)置 bossGroup 和 workerGroup 的, 為什么這里就只有一個(gè) bossGroup?

其實(shí)很簡(jiǎn)單, ServerBootstrap 重寫了 group 方法:

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

因此當(dāng)傳入一個(gè) group 時(shí), 那么 bossGroup 和 workerGroup 就是同一個(gè) NioEventLoopGroup 了. 并且這個(gè) NioEventLoopGroup 只有一個(gè)線程, 這樣就會(huì)導(dǎo)致 Netty 中的 acceptor 和后續(xù)的所有客戶端連接的 IO 操作都是在一個(gè)線程中處理的. 那么對(duì)應(yīng)到 Reactor 的線程模型中, 我們這樣設(shè)置 NioEventLoopGroup 時(shí), 就相當(dāng)于 Reactor 單線程模型.

同理, 再來(lái)看一下下面的例子:

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

bossGroup 中只有一個(gè)線程, 而 workerGroup 中的線程是 CPU 核心數(shù)乘以2, 因此對(duì)應(yīng)的到 Reactor 線程模型中, 我們知道, 這樣設(shè)置的 NioEventLoopGroup 其實(shí)就是 Reactor 多線程模型.

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

服務(wù)器端的 ServerSocketChannel 只綁定到了 bossGroup 中的一個(gè)線程, 因此在調(diào)用 Java NIO 的 Selector.select 處理客戶端的連接請(qǐng)求時(shí), 實(shí)際上是在一個(gè)線程中的, 所以對(duì)只有一個(gè)服務(wù)的應(yīng)用來(lái)說(shuō), bossGroup 設(shè)置多個(gè)線程是沒(méi)有什么作用的, 反而還會(huì)造成資源浪費(fèi).

bossGroup 是用于服務(wù)端的 accept, 即用于處理客戶端的連接請(qǐng)求. 我們可以把 Netty 比作一個(gè)飯店, bossGroup 就像一個(gè)像一個(gè)前臺(tái)接待, 當(dāng)客戶來(lái)到飯店吃時(shí), 接待員就會(huì)引導(dǎo)顧客就坐, 為顧客端茶送水等.

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

關(guān)于 bossGroup 與 workerGroup 的關(guān)系, 我們可以用如下圖來(lái)展示:

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

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

NioEventLoop -> SingleThreadEventLoop -> SingleThreadEventExecutor -> AbstractScheduledEventExecutor

在 AbstractScheduledEventExecutor 中, Netty 實(shí)現(xiàn)了 NioEventLoop 的 schedule 功能, 即我們可以通過(guò)調(diào)用一個(gè) NioEventLoop 實(shí)例的 schedule 方法來(lái)運(yùn)行一些定時(shí)任務(wù). 而在 SingleThreadEventLoop 中, 又實(shí)現(xiàn)了任務(wù)隊(duì)列的功能, 通過(guò)它, 我們可以調(diào)用一個(gè) NioEventLoop 實(shí)例的 execute 方法來(lái)向任務(wù)隊(duì)列中添加一個(gè) task, 并由 NioEventLoop 進(jìn)行調(diào)度執(zhí)行.

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

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

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

MultithreadEventExecutorGroup 中會(huì)調(diào)用 newChild 抽象方法來(lái)初始化 children 數(shù)組

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

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

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

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

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

當(dāng)一個(gè)任務(wù)被添加到 taskQueue 后, 它是怎么被 EventLoop 執(zhí)行的呢?
讓我們回到 NioEventLoop.run() 方法中, 在這個(gè)方法里, 會(huì)分別調(diào)用 processSelectedKeys() 和 runAllTasks() 方法, 來(lái)進(jìn)行 IO 事件的處理和 task 的處理.
runAllTasks 方法有兩個(gè)重載的方法, 一個(gè)是無(wú)參數(shù)的, 另一個(gè)有一個(gè)參數(shù)的. 首先來(lái)看一下無(wú)參數(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;
        }
    }
}

在此方法的一開始調(diào)用的 fetchFromScheduledTaskQueue() 其實(shí)就是將 scheduledTaskQueue 中已經(jīng)可以執(zhí)行的(即定時(shí)時(shí)間已到的 schedule 任務(wù)) 拿出來(lái)并添加到 taskQueue 中, 作為可執(zhí)行的 task 等待被調(diào)度執(zhí)行.

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

接下來(lái) runAllTasks() 方法就會(huì)不斷調(diào)用 task = pollTask() 從 taskQueue 中獲取一個(gè)可執(zhí)行的 task, 然后調(diào)用它的 run() 方法來(lái)運(yùn)行此 task.