《精通併發與Netty》學習筆記（11 - 詳解NIO (二) 分散/彙集 Scatter/Gather、Selector）

1、分散/彙集 Scatter/Gather

scatter/gather指的在多個緩衝區上實現一個簡單的I/O操做，好比從通道中讀取數據到多個緩衝區，或從多個緩衝區中寫入數據到通道；
scatter（分散）：指的是從通道中讀取數據分散到多個緩衝區Buffer的過程，該過程會將每一個緩存區填滿，直至通道中無數據或緩衝區沒有空間；
gather（彙集）：指的是將多個緩衝區Buffer彙集起來寫入到通道的過程，該過程相似於將多個緩衝區的內容鏈接起來寫入通道；

scatter/gather接口
以下是ScatteringByteChannel接口和GatheringByteChannel接口的定義，咱們能夠發現，接口中定義的方法都傳入了一個Buffer數組；
所謂的scatter/gather操做就是彙集（gather）這個Buffer數組並寫入到一個通道，或讀取通道數據並分散（scatter）到這個Buffer數組中；

public interface ScatteringByteChannel extends ReadableByteChannel
{
public long read(ByteBuffer[] dsts) throws IOException;

public long read(ByteBuffer[] dsts, int offset, int length) throws IOException;
}

public interface GatheringByteChannel extends WritableByteChannel
{
public long write(ByteBuffer[] srcs) throws IOException;

public long write(ByteBuffer[] srcs, int offset, int length) throws IOException;
}

提醒下，帶offset和length參數的read( ) 和write( )方法可讓咱們只使用緩衝區數組的子集，注意這裏的offset指的是緩衝區數組索引，而不是Buffer數據的索引，length指的是要使用的緩衝區數量；

以下代碼，將會往通道寫入第二個、第三個、第四個緩衝區內容；

int bytesRead = channel.write (fiveBuffers, 1, 3);

注意，不管是scatter仍是gather操做，都是按照buffer在數組中的順序來依次讀取或寫入的；

gather寫入
scatter / gather常常用於須要將傳輸的數據分開處理的場合，下面咱們看一下一個彙集寫入的示例：

ByteBuffer header = ByteBuffer.allocate(128);
ByteBuffer body = ByteBuffer.allocate(1024);
//write data into buffers
ByteBuffer[] bufferArray = { header, body };
channel.write(bufferArray);

以上代碼會將header和body這兩個緩衝區的數據寫入到通道；

這裏要特別注意的並非全部數據都寫入到通道，寫入的數據要根據position和limit的值來判斷，只有position和limit之間的數據纔會被寫入；
舉個例子，假如以上header緩衝區中有128個字節數據，但此時position=0，limit=58；那麼只有下標索引爲0-57的數據纔會被寫入到通道中；
scatter讀取
以下是一個分散讀取的示例：

ByteBuffer header = ByteBuffer.allocate(128);
ByteBuffer body = ByteBuffer.allocate(1024);
ByteBuffer[] bufferArray = { header, body };
channel.read(bufferArray);

以上代碼會將通道中的數據依次寫入到Buffer中，當一個buffer被寫滿後，channel緊接着向另外一個buffer中寫；

舉個例子，假如通道中有200個字節數據，那麼header會被寫入128個字節數據，body會被寫入72個字節數據；
好處
更加高效（如下內容摘自《JAVA NIO》）；
大多數現代操做系統都支持本地矢量I/O（native vectored I/O）操做。
當您在一個通道上請求一個Scatter/Gather操做時，該請求會被翻譯爲適當的本地調用來直接填充或抽取緩衝區，減小或避免了緩衝區拷貝和系統調用；
Scatter/Gather應該使用直接的ByteBuffers以從本地I/O獲取最大性能優點；

2、Selector

 

Selector（選擇器）是Java NIO中可以檢測一到多個NIO通道，並可以知曉通道是否爲諸如讀寫事件作好準備的組件。這樣，一個單獨的線程能夠管理多個channel，從而管理多個網絡鏈接。
爲何使用Selector?
僅用單個線程來處理多個Channels的好處是，只須要更少的線程來處理通道。事實上，能夠只用一個線程處理全部的通道。對於操做系統來講，線程之間上下文切換的開銷很大，並且每一個線程都要佔用系統的一些資源（如內存）。所以，使用的線程越少越好。

 

可是，須要記住，現代的操做系統和CPU在多任務方面表現的愈來愈好，因此多線程的開銷隨着時間的推移，變得愈來愈小了。實際上，若是一個CPU有多個內核，不使用多任務多是在浪費CPU能力。無論怎麼說，關於那種設計的討論應該放在另外一篇不一樣的文章中。在這裏，只要知道使用Selector可以處理多個通道就足夠了。

 

下面是單線程使用一個Selector處理3個channel的示例圖：

 

Selector的建立
經過調用Selector.open()方法建立一個Selector，以下：

Selector selector = Selector.open();

向Selector註冊通道
爲了將Channel和Selector配合使用，必須將channel註冊到selector上。經過SelectableChannel.register()方法來實現，以下：

channel.configureBlocking(false);
SelectionKey key = channel.register(selector,
    Selectionkey.OP_READ);

與Selector一塊兒使用時，Channel必須處於非阻塞模式下。這意味着不能將FileChannel與Selector一塊兒使用，由於FileChannel不能切換到非阻塞模式。而套接字通道均可以。

注意register()方法的第二個參數。這是一個「interest集合」，意思是在經過Selector監聽Channel時對什麼事件感興趣。能夠監聽四種不一樣類型的事件：

Connect
Accept
Read
Write
通道觸發了一個事件意思是該事件已經就緒。因此，某個channel成功鏈接到另外一個服務器稱爲「鏈接就緒」。一個server socket channel準備好接收新進入的鏈接稱爲「接收就緒」。一個有數據可讀的通道能夠說是「讀就緒」。等待寫數據的通道能夠說是「寫就緒」。

這四種事件用SelectionKey的四個常量來表示：

SelectionKey.OP_CONNECT
SelectionKey.OP_ACCEPT
SelectionKey.OP_READ
SelectionKey.OP_WRITE
若是你對不止一種事件感興趣，那麼能夠用「位或」操做符將常量鏈接起來，以下：

int interestSet = SelectionKey.OP_READ | SelectionKey.OP_WRITE;

在下面還會繼續提到interest集合。

SelectionKey
在上一小節中，當向Selector註冊Channel時，register()方法會返回一個SelectionKey對象。這個對象包含了一些你感興趣的屬性：

interest集合
ready集合
Channel
Selector
附加的對象（可選）
下面我會描述這些屬性。

interest集合
就像向Selector註冊通道一節中所描述的，interest集合是你所選擇的感興趣的事件集合。能夠經過SelectionKey讀寫interest集合，像這樣：

int interestSet = selectionKey.interestOps();
boolean isInterestedInAccept  = (interestSet & SelectionKey.OP_ACCEPT) == SelectionKey.OP_ACCEPT；
boolean isInterestedInConnect = interestSet & SelectionKey.OP_CONNECT;
boolean isInterestedInRead    = interestSet & SelectionKey.OP_READ;
boolean isInterestedInWrite   = interestSet & SelectionKey.OP_WRITE;

能夠看到，用「位與」操做interest 集合和給定的SelectionKey常量，能夠肯定某個肯定的事件是否在interest 集合中。

ready集合
ready 集合是通道已經準備就緒的操做的集合。在一次選擇(Selection)以後，你會首先訪問這個ready set。Selection將在下一小節進行解釋。能夠這樣訪問ready集合：

int readySet = selectionKey.readyOps();

能夠用像檢測interest集合那樣的方法，來檢測channel中什麼事件或操做已經就緒。可是，也可使用如下四個方法，它們都會返回一個布爾類型：

selectionKey.isAcceptable();
selectionKey.isConnectable();
selectionKey.isReadable();
selectionKey.isWritable();

Channel + Selector
從SelectionKey訪問Channel和Selector很簡單。以下：

Channel  channel  = selectionKey.channel();
Selector selector = selectionKey.selector();

附加的對象
能夠將一個對象或者更多信息附着到SelectionKey上，這樣就能方便的識別某個給定的通道。例如，能夠附加 與通道一塊兒使用的Buffer，或是包含彙集數據的某個對象。使用方法以下：

selectionKey.attach(theObject);
Object attachedObj = selectionKey.attachment();

還能夠在用register()方法向Selector註冊Channel的時候附加對象。如：

SelectionKey key = channel.register(selector, SelectionKey.OP_READ, theObject);

經過Selector選擇通道
一旦向Selector註冊了一或多個通道，就能夠調用幾個重載的select()方法。這些方法返回你所感興趣的事件（如鏈接、接受、讀或寫）已經準備就緒的那些通道。換句話說，若是你對「讀就緒」的通道感興趣，select()方法會返回讀事件已經就緒的那些通道。

下面是select()方法：

• int select()
• int select(long timeout)
• int selectNow()

select()阻塞到至少有一個通道在你註冊的事件上就緒了。

select(long timeout)和select()同樣，除了最長會阻塞timeout毫秒(參數)。

selectNow()不會阻塞，無論什麼通道就緒都馬上返回（譯者注：此方法執行非阻塞的選擇操做。若是自從前一次選擇操做後，沒有通道變成可選擇的，則此方法直接返回零。）。

select()方法返回的int值表示有多少通道已經就緒。亦即，自上次調用select()方法後有多少通道變成就緒狀態。若是調用select()方法，由於有一個通道變成就緒狀態，返回了1，若再次調用select()方法，若是另外一個通道就緒了，它會再次返回1。若是對第一個就緒的channel沒有作任何操做，如今就有兩個就緒的通道，但在每次select()方法調用之間，只有一個通道就緒了。

selectedKeys()
一旦調用了select()方法，而且返回值代表有一個或更多個通道就緒了，而後能夠經過調用selector的selectedKeys()方法，訪問「已選擇鍵集（selected key set）」中的就緒通道。以下所示：

Set selectedKeys = selector.selectedKeys();

當像Selector註冊Channel時，Channel.register()方法會返回一個SelectionKey 對象。這個對象表明了註冊到該Selector的通道。能夠經過SelectionKey的selectedKeySet()方法訪問這些對象。
能夠遍歷這個已選擇的鍵集合來訪問就緒的通道。以下：

Set selectedKeys = selector.selectedKeys();
Iterator keyIterator = selectedKeys.iterator();
while(keyIterator.hasNext()) {
    SelectionKey key = keyIterator.next();
    if(key.isAcceptable()) {
        // a connection was accepted by a ServerSocketChannel.
    } else if (key.isConnectable()) {
        // a connection was established with a remote server.
    } else if (key.isReadable()) {
        // a channel is ready for reading
    } else if (key.isWritable()) {
        // a channel is ready for writing
    }
    keyIterator.remove();
}

這個循環遍歷已選擇鍵集中的每一個鍵，並檢測各個鍵所對應的通道的就緒事件。

注意每次迭代末尾的keyIterator.remove()調用。Selector不會本身從已選擇鍵集中移除SelectionKey實例。必須在處理完通道時本身移除。下次該通道變成就緒時，Selector會再次將其放入已選擇鍵集中。

SelectionKey.channel()方法返回的通道須要轉型成你要處理的類型，如ServerSocketChannel或SocketChannel等。

wakeUp()
某個線程調用select()方法後阻塞了，即便沒有通道已經就緒，也有辦法讓其從select()方法返回。只要讓其它線程在第一個線程調用select()方法的那個對象上調用Selector.wakeup()方法便可。阻塞在select()方法上的線程會立馬返回。

若是有其它線程調用了wakeup()方法，但當前沒有線程阻塞在select()方法上，下個調用select()方法的線程會當即「醒來（wake up）」。

close()
用完Selector後調用其close()方法會關閉該Selector，且使註冊到該Selector上的全部SelectionKey實例無效。通道自己並不會關閉。

完整的示例
這裏有一個完整的示例，打開一個Selector，註冊一個通道註冊到這個Selector上(通道的初始化過程略去),而後持續監控這個Selector的四種事件（接受，鏈接，讀，寫）是否就緒。

Selector selector = Selector.open();
channel.configureBlocking(false);
SelectionKey key = channel.register(selector, SelectionKey.OP_READ);
while(true) {
  int readyChannels = selector.select();
  if(readyChannels == 0) continue;
  Set selectedKeys = selector.selectedKeys();
  Iterator keyIterator = selectedKeys.iterator();
  while(keyIterator.hasNext()) {
    SelectionKey key = keyIterator.next();
    if(key.isAcceptable()) {

        // a connection was accepted by a ServerSocketChannel.
    } else if (key.isConnectable()) {
        // a connection was established with a remote server.
    } else if (key.isReadable()) {
        // a channel is ready for reading
    } else if (key.isWritable()) {
        // a channel is ready for writing
    }
    keyIterator.remove();

  }
}