編者注：Netty是Java領域有名的開源網路庫，特點是高效能和高擴充套件性，因此很多流行的框架都是基於它來構建的，比如我們熟知的Dubbo、Rocketmq、Hadoop等。本文就netty執行緒模型展開分析討論下 : )

IO模型

• BIO：同步阻塞IO模型；
• NIO：基於IO多路複用技術的“非阻塞同步”IO模型。簡單來說，核心將可讀可寫事件通知應用，由應用主動發起讀寫操作；
• AIO：非阻塞非同步IO模型。簡單來說，核心將讀完成事件通知應用，讀操作由核心完成，應用只需操作資料即可；應用做非同步寫操作時立即返回，核心會進行寫操作排隊並執行寫操作。

NIO和AIO不同之處在於應用是否進行真正的讀寫操作。

reactor和proactor模型

• reactor：基於NIO技術，可讀可寫時通知應用；
• proactor：基於AIO技術，讀完成時通知應用，寫操作應用通知核心。

netty執行緒模型

netty的執行緒模型是基於Reactor模型的。

netty單執行緒模型

Reactor 單執行緒模型，是指所有的 I/O 操作都在同一個 NIO 執行緒上面完成的，此時NIO執行緒職責包括：接收新建連線請求、讀寫操作等。

在一些小容量應用場景下，可以使用單執行緒模型（注意，Redis的請求處理也是單執行緒模型，為什麼Redis的效能會如此之高呢？因為Redis的讀寫操作基本都是記憶體操作，並且Redis協議比較簡潔，序列化/反序列化耗費效能更低

）。但是對於高負載、大併發的應用場景卻不合適，主要原因如下：

• 一個NIO執行緒同時處理成百上千的連線，效能上無法支撐，即便NIO執行緒的CPU負荷達到100%，也無法滿足海量訊息的編碼、解碼、讀取和傳送。
• 當NIO執行緒負載過重之後，處理速度將變慢，這會導致大量客戶端連線超時，超時之後往往會進行重發，這更加重了NIO執行緒的負載，最終會導致大量訊息積壓和處理超時，成為系統的效能瓶頸。
• 可靠性問題：一旦NIO執行緒意外跑飛，或者進入死迴圈，會導致整個系統通訊模組不可用，不能接收和處理外部訊息，造成節點故障。

Reactor多執行緒模型

Rector 多執行緒模型與單執行緒模型最大的區別就是有一組 NIO 執行緒來處理連線讀寫操作，一個NIO執行緒處理Accept。一個NIO執行緒可以處理多個連線事件，一個連線的事件只能屬於一個NIO執行緒。

在絕大多數場景下，Reactor 多執行緒模型可以滿足效能需求。但是，在個別特殊場景中，一個 NIO 執行緒負責監聽和處理所有的客戶端連線可能會存在效能問題。例如併發百萬客戶端連線，或者服務端需要對客戶端握手進行安全認證，但是認證本身非常損耗效能。在這類場景下，單獨一個 Acceptor 執行緒可能會存在效能不足的問題，為了解決效能問題，產生了第三種 Reactor 執行緒模型——主從Reactor 多執行緒模型。

Reactor主從多執行緒模型

主從 Reactor 執行緒模型的特點是：服務端用於接收客戶端連線的不再是一個單獨的 NIO 執行緒，而是一個獨立的 NIO 執行緒池。Acceptor 接收到客戶端 TCP連線請求並處理完成後（可能包含接入認證等），將新建立的 SocketChannel注 冊 到 I/O 線 程 池（sub reactor 線 程 池）的某個I/O執行緒上， 由它負責SocketChannel 的讀寫和編解碼工作。Acceptor 執行緒池僅僅用於客戶端的登入、握手和安全認證，一旦鏈路建立成功，就將鏈路註冊到後端 subReactor 執行緒池的 I/O 執行緒上，由 I/O 執行緒負責後續的 I/O 操作。

netty執行緒模型思考

netty 的執行緒模型並不是一成不變的，它實際取決於使用者的啟動引數配置。通過設定不同的啟動引數，Netty 可以同時支援 Reactor 單執行緒模型、多執行緒模型。

為了儘可能地提升效能，Netty 在很多地方進行了無鎖化的設計，例如在 I/O 執行緒內部進行序列操作，避免多執行緒競爭導致的效能下降問題。表面上看，序列化設計似乎 CPU 利用率不高，併發程度不夠。但是，通過調整 NIO 執行緒池的執行緒引數，可以同時啟動多個序列化的執行緒並行執行，這種區域性無鎖化的序列執行緒設計相比一個佇列多個工作執行緒的模型效能更優。（小夥伴們後續多執行緒併發流程可參考該類實現方案）

Netty 的 NioEventLoop 讀取到訊息之後，直接呼叫 ChannelPipeline 的fireChannelRead (Object msg)。 只要使用者不主動切換執行緒， 一直都是由NioEventLoop 呼叫使用者的 ChannelHandler，期間不進行執行緒切換。這種序列化處理方式避免了多執行緒操作導致的鎖的競爭，從效能角度看是最優的。

Netty擁有兩個NIO執行緒池，分別是bossGroup和workerGroup，前者處理新建連線請求，然後將新建立的連線輪詢交給workerGroup中的其中一個NioEventLoop來處理，後續該連線上的讀寫操作都是由同一個NioEventLoop來處理。注意，雖然bossGroup也能指定多個NioEventLoop（一個NioEventLoop對應一個執行緒），但是預設情況下只會有一個執行緒，因為一般情況下應用程式只會使用一個對外監聽埠。

這裡試想一下，難道不能使用多執行緒來監聽同一個對外埠麼，即多執行緒epoll_wait到同一個epoll例項上？

epoll相關的主要兩個方法是epoll_wait和epoll_ctl，多執行緒同時操作同一個epoll例項，那麼首先需要確認epoll相關方法是否執行緒安全：簡單來說，epoll是通過鎖來保證執行緒安全的, epoll中粒度最小的自旋鎖ep->lock(spinlock)用來保護就緒的佇列, 互斥鎖ep->mtx用來保護epoll的重要資料結構紅黑樹。

看到這裡，可能有的小夥伴想到了Nginx多程序針對監聽埠的處理策略，Nginx是通過accept_mutex機制來保證的。accept_mutex是nginx的(新建連線)負載均衡鎖，讓多個worker程序輪流處理與client的新連線。當某個worker程序的連線數達到worker_connections配置（單個worker程序的最大處理連線數）的最大連線數的7/8時，會大大減小獲取該worker獲取accept鎖的概率，以此實現各worker程序間的連線數的負載均衡。accept鎖預設開啟，關閉它時nginx處理新建連線耗時會更短，但是worker程序之間可能連線不均衡，並且存在“驚群”問題。只有在使能accept_mutex並且當前系統不支援原子鎖時，才會用檔案實現accept鎖。注意，accept_mutex加鎖失敗時不會阻塞當前執行緒，類似tryLock。

現代linux中，多個socker同時監聽同一個埠也是可行的，nginx 1.9.1也支援這一行為。linux 3.9以上核心支援SO_REUSEPORT選項，允許多個socker bind/listen在同一埠上。這樣，多個程序可以各自申請socker監聽同一埠，當連線事件來臨時，核心做負載均衡，喚醒監聽的其中一個程序來處理，reuseport機制有效的解決了epoll驚群問題。

再回到剛才提出的問題，java中多執行緒來監聽同一個對外埠，epoll方法是執行緒安全的，這樣就可以使用使用多執行緒監聽epoll_wait了麼，當然是不建議這樣乾的，除了epoll的驚群問題之外，還有一個就是，一般開發中我們使用epoll設定的是LT模式（水平觸發方式，與之相對的是ET預設，前者只要連線事件未被處理就會在epoll_wait時始終觸發，後者只會在真正有事件來時在epoll_wait觸發一次），這樣的話，多執行緒epoll_wait時就會導致第一個執行緒epoll_wait之後還未處理完畢已發生的事件時，第二個執行緒也會epoll_wait返回，顯然這不是我們想要的，關於java nio的測試demo如下：

public class NioDemo {
    private static AtomicBoolean flag = new AtomicBoolean(true);
    public static void main(String[] args) throws Exception {
        ServerSocketChannel serverChannel = ServerSocketChannel.open();
        serverChannel.socket().bind(new InetSocketAddress(8080));
        // non-block io
        serverChannel.configureBlocking(false);
        Selector selector = Selector.open();
        serverChannel.register(selector, SelectionKey.OP_ACCEPT);

        // 多執行緒執行
        Runnable task = () -> {
            try {
                while (true) {
                    if (selector.select(0) == 0) {
                        System.out.println("selector.select loop... " + Thread.currentThread().getName());
                        Thread.sleep(1);
                        continue;
                    }

                    if (flag.compareAndSet(true, false)) {
                        System.out.println(Thread.currentThread().getName() + " over");
                        return;
                    }

                    Iterator<SelectionKey> iter = selector.selectedKeys().iterator();
                    while (iter.hasNext()) {
                        SelectionKey key = iter.next();

                        // accept event
                        if (key.isAcceptable()) {
                            handlerAccept(selector, key);
                        }

                        // socket event
                        if (key.isReadable()) {
                            handlerRead(key);
                        }

                        /**
                         * Selector不會自己從已選擇鍵集中移除SelectionKey例項，必須在處理完通道時手動移除。
                         * 下次該通道變成就緒時，Selector會再次將其放入已選擇鍵集中。
                         */
                        iter.remove();
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        };

        List<Thread> threadList = new ArrayList<>();
        for (int i = 0; i < 2; i++) {
            Thread thread = new Thread(task);
            threadList.add(thread);
            thread.start();
        }
        for (Thread thread : threadList) {
            thread.join();
        }
        System.out.println("main end");
    }

    static void handlerAccept(Selector selector, SelectionKey key) throws Exception {
        System.out.println("coming a new client... " + Thread.currentThread().getName());
        Thread.sleep(10000);
        SocketChannel channel = ((ServerSocketChannel) key.channel()).accept();
        channel.configureBlocking(false);
        channel.register(selector, SelectionKey.OP_READ, ByteBuffer.allocate(1024));
    }

    static void handlerRead(SelectionKey key) throws Exception {
        SocketChannel channel = (SocketChannel) key.channel();
        ByteBuffer buffer = (ByteBuffer) key.attachment();
        buffer.clear();

        int num = channel.read(buffer);
        if (num <= 0) {
            // error or fin
            System.out.println("close " + channel.getRemoteAddress());
            channel.close();
        } else {
            buffer.flip();
            String recv = Charset.forName("UTF-8").newDecoder().decode(buffer).toString();
            System.out.println("recv: " + recv);

            buffer = ByteBuffer.wrap(("server: " + recv).getBytes());
            channel.write(buffer);
        }
    }
}

netty執行緒模型實踐

(1) 時間可控的簡單業務直接在 I/O 執行緒上處理

時間可控的簡單業務直接在 I/O 執行緒上處理，如果業務非常簡單，執行時間非常短，不需要與外部網路互動、訪問資料庫和磁碟，不需要等待其它資源，則建議直接在業務 ChannelHandler 中執行，不需要再啟業務的執行緒或者執行緒池。避免執行緒上下文切換，也不存線上程併發問題。

(2) 複雜和時間不可控業務建議投遞到後端業務執行緒池統一處理

複雜度較高或者時間不可控業務建議投遞到後端業務執行緒池統一處理，對於此類業務，不建議直接在業務 ChannelHandler 中啟動執行緒或者執行緒池處理，建議將不同的業務統一封裝成 Task，統一投遞到後端的業務執行緒池中進行處理。過多的業務ChannelHandler 會帶來開發效率和可維護性問題，不要把 Netty 當作業務容器，對於大多數複雜的業務產品，仍然需要整合或者開發自己的業務容器，做好和Netty 的架構分層。

(3) 業務執行緒避免直接操作 ChannelHandler

業務執行緒避免直接操作 ChannelHandler，對於 ChannelHandler，IO 執行緒和業務執行緒都可能會操作，因為業務通常是多執行緒模型，這樣就會存在多執行緒操作ChannelHandler。為了儘量避免多執行緒併發問題，建議按照 Netty 自身的做法，通過將操作封裝成獨立的 Task 由 NioEventLoop 統一執行，而不是業務執行緒直接操作，相關程式碼如下所示：

如果你確認併發訪問的資料或者併發操作是安全的，則無需多此一舉，這個需要根據具體的業務場景進行判斷，靈活處理。

推薦閱讀

• Java nio 空輪詢bug到底是什麼
• Netty 入門，這一篇文章就夠了
• Netty 啟動流程解析
• Netty 處理連線那些事
• Java常見幾種動態代理的對比
• 程式設計師必看| mockito原理淺析
• Eureka 原理分析
• MQ初窺門徑【面試必看的Kafka和RocketMQ儲存區別】
• java lambda 深入淺出

歡迎小夥伴關注【TopCoder】閱讀更多精彩好文。

相關推薦