Java NIO入門學習（三）
阿新 • • 發佈：2018-12-25
在上一篇中介紹了緩衝區的原理，下面來介紹NIO中另一個核心物件選擇器（Selector）以及NIO的原理。
在Client/Server模型中，Server往往需要同時處理大量來自Client的訪問請求，因此Server端需採用支援高併發訪問的架構。一種簡單而又直接的解決方案是“one-thread-per-connection”。這是一種基於阻塞式I/O的多執行緒模型。在該模型中，Server為每個Client連線建立一個處理執行緒，每個處理執行緒阻塞式等待可能達到的資料，一旦資料到達，則立即處理請求、返回處理結果並再次進入等待狀態。由於每個Client連線有一個單獨的處理執行緒為其服務，因此可保證良好的響應時間。但當系統負載增大（併發請求增多）時，Server端需要的執行緒數會增加，這將成為系統擴充套件的瓶頸所在。
Java NIO不但引入了全新的高效的I/O機制，同時引入了基於Reactor設計模式的多路複用非同步模式。NIO包中主要包含以下幾種核心物件。
* Channel(通道)：NIO把它支援的I/O物件抽象為Channel。它模擬了通訊連線，類似於原I/O中的流（Stream），使用者可以通過它讀取和寫入資料。例項類有SocketChannel、ServerSocketChannel、DatagramChannel、FileChannel等。
* Buffer(緩衝區)：Buffer是一塊連續的記憶體區域，一般作為Channel收發資料的載體出現。所有資料都通過Buffer物件來處理。
* Selector(選擇器)：Selector類提供了監控一個和多個通道當前狀態的機制。只要Channel向Selector註冊了某種特定的事件，Selector就會監聽這些事件是否會發生，一旦發生某個事件，便會通知對應的Channel。使用選擇器，藉助單一執行緒，就可對數量龐大的活動I/O通道實施監控和維護。
Java NIO的服務端只需啟動一個專門的執行緒來處理所有的IO事件，這種通訊模型是怎麼實現的呢？Java NIO採用了雙向通道（Channel）進行資料傳輸，而不是單向的流（Stream），在通道上可以註冊我們感興趣的事件。一共有以下四種事件：
| 事件名 |
對應值 |
| 服務端接收客戶端連線事件 |
SelectionKey.OP_ACCEPT(16) |
| 客戶端連線服務端事件 |
SelectionKey.OP_CONNECT(8) |
| 讀事件 |
SelectionKey.OP_READ(1) |
| 寫事件 |
SelectionKey.OP_WRITE(4) |
服務端和客戶端各自維護一個管理通道的物件，我們稱之為Selector，該物件能檢測一個或多個通道 (Channel) 上的事件。我們以服務端為例，如果服務端的Selector上註冊了讀事件，某時刻客戶端給服務端傳送了一些資料，阻塞I/O這時會呼叫read()方法阻塞地讀取資料，而NIO的服務端會在Selector中註冊一個讀事件，然後服務端的處理執行緒會輪詢地訪問Selector，如果發現有讀事件到達，則處理這些事件，如果沒有則處理執行緒會一直阻塞直到讀事件到達為止。
為了更好地理解java NIO，下面貼出服務端和客戶端的簡單程式碼實現。
服務端：import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
/**
* NIO服務端
*
*/
public class NIOServer {
private Selector selector;
private ByteBuffer echoBuffer = ByteBuffer.allocate(1024);
/**
* 獲得一個ServerSocket通道，並對該通道做一些初始化的工作
*
* @param port
* @throws IOException
*/
public void initServer(int port) throws IOException {
selector = Selector.open();
// 獲得一個ServerSocket通道，並設定為非阻塞
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.configureBlocking(false);
// 將通道對應的ServerSocket繫結到port埠
ServerSocket serverSocket = serverSocketChannel.socket();
InetSocketAddress address = new InetSocketAddress(port);
serverSocket.bind(address);
// 將通道管理器和該通道繫結，併為該通道註冊SelectionKey.OP_ACCEPT事件。
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
}
/**
* 迴圈監聽selector上是否有需要處理的事件，如果有，則進行處理
*
* @throws IOException
*/
public void listen() throws IOException {
System.out.println("服務端啟動成功！");
// 迴圈訪問selector
while (true) {
// //當註冊的事件到達時，方法返回；否則,該方法會一直阻塞
selector.select();
// 返回發生了事件的SelectionKey物件的一個 集合
Set<SelectionKey> selectedKeys = selector.selectedKeys();
Iterator<SelectionKey> it = selectedKeys.iterator();
while (it.hasNext()) {
SelectionKey key = (SelectionKey) it.next();
it.remove(); // 刪除已選的key以防重複處理
// 客戶端請求連線事件
if ((key.readyOps() & SelectionKey.OP_ACCEPT) == SelectionKey.OP_ACCEPT) {
ServerSocketChannel server = (ServerSocketChannel) key
.channel();
SocketChannel channel = server.accept(); // 獲得和客戶端連線的通道
channel.configureBlocking(false);// 設定成非阻塞
// 在和客戶端連線成功之後，為了可以接收到客戶端的資訊，需要給通道註冊讀事件。
channel.register(this.selector, SelectionKey.OP_READ);
} else if ((key.readyOps() & SelectionKey.OP_READ) == SelectionKey.OP_READ) {
// 得到可讀事件發生的Socket通道
SocketChannel channel = (SocketChannel) key.channel();
while (true) {
echoBuffer.clear();
int r = channel.read(echoBuffer);
// 判斷拷貝是否完成
if (r <= 0) {
break;
}
echoBuffer.flip();
channel.write(echoBuffer); // 將訊息回送給客戶端
}
byte[] data = echoBuffer.array();
System.out.println("伺服器收到並返回資訊：" + new String(data).trim());
}
}
}
}
/**
* 啟動服務端測試
*
* @throws IOException
*/
public static void main(String[] args) throws IOException {
NIOServer server = new NIOServer();
server.initServer(8000);
server.listen();
}
}
客戶端：
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
/**
* NIO客戶端
*
*/
public class NIOClient {
private Selector selector;
/**
* 獲得一個Socket通道，並對該通道做一些初始化的工作
*
* @param ip連線的伺服器的ip
* @param port連線的伺服器的埠號
* @throws IOException
*/
public void initClient(String ip, int port) throws IOException {
selector = Selector.open();
// 獲得一個Socket通道，並設定為非阻塞
SocketChannel channel = SocketChannel.open();
channel.configureBlocking(false);
// 客戶端連線伺服器,其實方法執行並沒有實現連線，需要在listen()方法中調
// 用channel.finishConnect();才能完成連線
channel.connect(new InetSocketAddress(ip, port));
// 將通道管理器和該通道繫結，併為該通道註冊SelectionKey.OP_CONNECT事件。
channel.register(selector, SelectionKey.OP_CONNECT);
}
/**
* 迴圈監聽selector上是否有需要處理的事件，如果有，則進行處理
*
* @throws IOException
*/
public void listen() throws IOException {
// 迴圈訪問selector
while (true) {
// //當註冊的事件到達時，方法返回；否則,該方法會一直阻塞
selector.select();
// 返回發生了事件的SelectionKey物件的一個 集合
Set<SelectionKey> selectedKeys = selector.selectedKeys();
Iterator<SelectionKey> it = selectedKeys.iterator();
while (it.hasNext()) {
SelectionKey key = (SelectionKey) it.next();
it.remove(); // 刪除已選的key以防重複處理
// 連線事件發生
if ((key.readyOps() & SelectionKey.OP_CONNECT) == SelectionKey.OP_CONNECT) {
// 獲得和伺服器連線的通道
SocketChannel channel = (SocketChannel) key.channel();
// 如果正在連線，則完成連線
if (channel.isConnectionPending()) {
channel.finishConnect();
}
channel.configureBlocking(false);// 設定成非阻塞
// 在和伺服器連線成功之後，為了可以向伺服器傳送資訊，需要為通道註冊寫事件。
channel.register(this.selector, SelectionKey.OP_WRITE);
} else if ((key.readyOps() & SelectionKey.OP_WRITE) == SelectionKey.OP_WRITE) {
// 得到可寫事件發生的Socket通道
SocketChannel channel = (SocketChannel) key.channel();
String msg = new String("Hello, Nice to meet you!");
channel.write(ByteBuffer.wrap(msg.getBytes())); // 向伺服器傳送資訊
// 在向伺服器寫完資訊之後，為了可以接收到伺服器返回的資訊，需要給通道註冊讀事件。
channel.register(this.selector, SelectionKey.OP_READ);
System.out.println("客戶端傳送資訊：" + msg);
} else if ((key.readyOps() & SelectionKey.OP_READ) == SelectionKey.OP_READ) {
// 得到可讀事件發生的Socket通道
SocketChannel channel = (SocketChannel) key.channel();
ByteBuffer echoBuffer = ByteBuffer.allocate(1024);
while (true) {
echoBuffer.clear();
int r = channel.read(echoBuffer);
// 判斷拷貝是否完成
if (r <= 0) {
break;
}
}
byte[] data = echoBuffer.array();
System.out.println("客戶端收到資訊：" + new String(data).trim());
}
}
}
}
/**
* 啟動客戶端測試
*
* @throws IOException
*/
public static void main(String[] args) throws IOException {
NIOClient client = new NIOClient();
client.initClient("localhost", 8000);
client.listen();
}
}
輸出結果為：
服務端：
客戶端：
參考：