Netty多線程處理機制

技術點描述

本文主要研究NioServerSocketChannelFactory類和NioDatagramChannelFactory類,

以及這兩個類的各自作用。

    由於基於pipelineFactory的線程池的配置已在前一節（Netty中execution包功能詳解 ）中做了詳細說明，而channelFactory中的線程池的配置並未做說明，本文檔重點闡述channelFactory的線程池配置。NioServerSocketChannelFactory主要應用於TCP協議的數據處理，NioDatagramChannelFactory主要應用於UDP協議的數據處理。

實現方案

簡單介紹一下實現該功能，使用了哪些技術或知識點

參考源碼包

以下是對這兩個類的具體說明以及相關類，重要的常用的方法的說明

1. NioServerSocketChannelFactory

此類常用於創建TCP的ServerBootstrap時作為構造方法的參數使用。

調用方式及設置如下：

1 ChannelFactory TCPCHANNEL_FACTORY = new NioServerSocketChannelFactory(
2 
3 Executors.newCachedThreadPool(), Executors.newCachedThreadPool());
4 
5 ServerBootstrap TCPSERVER_BOOTSTRAP = new ServerBootstrap(TCPCHANNEL_FACTORY);

基於NIO的用於創建服務器端channel的channelFactory，這個類利用非阻塞I/O模式引入NIO，這樣可以有效地向多個並發連接提供Channel

線程的工作模式：

在此類中，有兩種類型的線程，一種是boss線程，另一種是worker線程

 

Boss線程：

每個server服務器都會有一個boss線程，每綁定一個InetSocketAddress都會產生一個boss線程，比如：我們開啟了兩個服務器端口80和443，則我們會有兩個boss線程。一個boss線程在端口綁定后，會接收傳進來的連接，一旦連接接收成功，boss線程會指派一個worker線程處理連接。

Worker線程：

一個NioServerSocketChannelFactory會有一個或者多個worker線程。一個worker線程在非阻塞模式下為一個或多個Channels提供非阻塞 讀或寫

線程的生命周期和優雅的關閉

在NioServerSocketChannelFactory被創建的時候，所有的線程都會從指定的Executors中獲取。Boss線程從bossExecutor中獲取，worker線程從workerExecutor中獲取。因此，我們應該准確的指定Executors可以提供足夠數量的線程，最好的選擇就是指定一個cached線程池（It is the best bet to specify a cached thread pool）。

此處發現所有源碼中的例子(example)中均設置為Executors.newCachedThreadPool()

Boss線程和worker線程都是懶加載，沒有程序使用的時候要釋放掉。當boss線程和worker線程釋放掉的時候，所有的相關資源如Selector也要釋放掉。因此，如果想要優雅的關閉一個服務，需要做一下事情：

1. 對factory創建的channels執行解綁（unbind）操作
2. 關閉所有的由解綁的channels處理的子channels（這兩步目前通常通過ChannelGroup.close()來操作）
3. 調用releaseExternalResources()方法

請確保在所有的channels都關閉前不要關閉executor，否則，會報RejectedExecutionException異常而且相關資源可能不會被釋放掉。

 

在此類中，最關鍵的構造方法：

 1 public NioServerSocketChannelFactory(
 2 
 3 Executor bossExecutor, WorkerPool<NioWorker> workerPool) {
 4 
 5 if (bossExecutor == null) {
 6 
 7 throw new NullPointerException("bossExecutor");
 8 
 9 }
10 
11 if (workerPool == null) {
12 
13 throw new NullPointerException("workerPool");
14 
15 }
16 
17  
18 
19 this.bossExecutor = bossExecutor;
20 
21 this.workerPool = workerPool;
22 
23 sink = new NioServerSocketPipelineSink(workerPool);
24 
25 }

參數說明：

bossExecutor:默認推薦設置為Executors.newCachedThreadPool()

workerPool：可通過new NioWorkerPool(workerExecutor, workerCount, true)創建，workerExecutor默認推薦設置為Executors.newCachedThreadPool()，workerCount可設置為Runtime.getRuntime().availableProcessors() * 2（默認）

關於其中的NioWorkerPool類和NioServerSocketPipelineSink類，將在下文給予說明。

1. NioWorkerPool

此類的關鍵方法是super的AbstractNioWorkerPool，

構造方法：

1 AbstractNioWorkerPool(Executor workerExecutor, int workerCount, boolean allowShutDownOnIdle) {
2 
3 …
4 
5 }

參數說明：

workerExecutor：worker線程的Executor

workerCount：最多創建的worker線程數

allowShutDownOnldle：空閑時是否關閉線程

1. NioServerSocketPipelineSink

內部組件，傳輸服務的一個實現類，大多數情況下，不對用戶開放。

1. NioDatagramChannelFactory

此類常用於創建UDP的ServerBootstrap時作為構造方法的參數使用。

調用方式及設置如下：

1 ChannelFactory UDPCHANNEL_FACTORY = new NioDatagramChannelFactory()
2 
3 ConnectionlessBootstrap UDPSERVER_BOOTSTRAP = new ConnectionlessBootstrap(UDPCHANNEL_FACTORY);

這個類利用非阻塞I/O模式引入NIO，這樣可以有效地向多個並發連接提供Channel

此類僅有一種線程類型：worker線程。一個NioDatagramChannelFactory可創建一個或多個worker線程，一個worker線程在非阻塞模式下為一個或多個Channels提供非阻塞 讀或寫

在NioDatagramChannelFactory被創建的時候，所有的線程都會從指定的Executors中獲取

因此，我們應該准確的指定Executors可以提供足夠數量的線程，最好的選擇就是指定一個cached線程池（It is the best bet to specify a cached thread pool）。

所有的worker線程都是懶加載，沒有程序使用的時候要釋放掉。當worker線程釋放掉的時候，所有的相關資源如Selector也要釋放掉。因此，如果想要優雅的關閉一個服務，需要做一下事情：

通常通過ChannelGroup.close()來關閉所有的由此factory創建的channels

調用releaseExternalResources()方法

請確保在所有的channels都關閉前不要關閉executor，否則，會報RejectedExecutionException異常而且相關資源可能不會被釋放掉。

 

不支持多播模式，如果需要多播模式支持，采用OioDatagramChannelFactory替代

以下是此類中的最重要的構造方法：

public NioDatagramChannelFactory(WorkerPool<NioDatagramWorker> workerPool, InternetProtocolFamily family) {

…

}

參數說明：

workerPool：參考NioServerSocketChannelFactory

family：網絡協議系列，這個參數是為了UDP的多播模式的，此參數只在java7+版本才有效。默認為null，可通過StandardProtocolFamily.INET(Ipv4)或者StandardProtocolFamily.INET6(Ipv6)設置