一、阻塞IO與非阻塞IO
阻塞IO:
通常在進行同步I/O操作時，如果讀取數據，代碼會阻塞直至有 可供讀取的數據。同樣，寫入調用將會阻塞直至數據能夠寫入。傳統的Server/Client模式會基於TPR（Thread per Request）,服務器會為每個客戶端請求建立一個線程，由該線程單獨負責處理一個客戶請求。這種模式帶來的一個問題就是線程數量的劇增，大量的線程會增大服務器的開銷。大多數的實現為了避免這個問題，都采用了線程池模型，並設置線程池線程的最大數量，這由帶來了新的問題，如果線程池中有200個線程，而有200個用戶都在進行大文件下載，會導致第201個用戶的請求無法及時處理，即便第201個用戶只想請求一個幾KB大小的頁面。傳統的 Server/Client模式如下圖所示：
非阻塞IO(NIO):
NIO中非阻塞I/O采用了基於Reactor模式的工作方式，I/O調用不會被阻塞，相反是注冊感興趣的特定I/O事件，如可讀數據到達，新的套接字連接等等，在發生特定事件時，系統再通知我們。NIO中實現非阻塞I/O的核心對象就是Selector，Selector就是注冊各種I/O事件地 方，而且當那些事件發生時，就是這個對象告訴我們所發生的事件，如下圖所示：
從圖中可以看出，當有讀或寫等任何注冊的事件發生時，可以從Selector中獲得相應的SelectionKey，同時從 SelectionKey中可以找到發生的事件和該事件所發生的具體的SelectableChannel，以獲得客戶端發送過來的數據。
非阻塞指的是IO事件本身不阻塞,但是獲取IO事件的select()方法是需要阻塞等待的.區別是阻塞的IO會阻塞在IO操作上, NIO阻塞在事件獲取上,沒有事件就沒有IO, 從高層次看IO就不阻塞了.也就是說只有IO已經發生那么我們才評估IO是否阻塞,但是select()阻塞的時候IO還沒有發生,何談IO的阻塞呢?NIO的本質是延遲IO操作到真正發生IO的時候,而不是以前的只要IO流打開了就一直等待IO操作。
二.NIO原理及通信模型
Java NIO是在jdk1.4開始使用的，它既可以說成“新I/O”，也可以說成非阻塞式I/O。下面是java NIO的工作原理：
1. 由一個專門的線程來處理所有的 IO 事件，並負責分發。
2. 事件驅動機制：事件到的時候觸發，而不是同步的去監視事件。
3. 線程通訊：線程之間通過 wait,notify 等方式通訊。保證每次上下文切換都是有意義的。減少無謂的線程切換。
閱讀過一些資料之后，下面貼出我理解的java NIO的工作原理圖：
（注：每個線程的處理流程大概都是讀取數據、解碼、計算處理、編碼、發送響應。）
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，如果訪問selector時發現有感興趣的事件到達，則處理這些事件，如果沒有感興趣的事件到達，則處理線程會一直阻塞直到感興趣的事件到達為止。下面是我理解的java NIO的通信模型示意圖：
三、Java NIO 概述
Java NIO 由以下幾個核心部分組成：
- Channels
- Buffers
- Selectors
雖然Java NIO 中除此之外還有很多類和組件，但在我看來，Channel，Buffer 和 Selector 構成了核心的API。其它組件，如Pipe和FileLock，只不過是與三個核心組件共同使用的工具類。因此，在概述中我將集中在這三個組件上。其它組件會在單獨的章節中講到。
Channel 和 Buffer
基本上，所有的 IO 在NIO 中都從一個Channel 開始。Channel 有點象流。 數據可以從Channel讀到Buffer中，也可以從Buffer 寫到Channel中。這里有個圖示：
Channel和Buffer有好幾種類型。下面是JAVA NIO中的一些主要Channel的實現：
- FileChannel
- DatagramChannel
- SocketChannel
- ServerSocketChannel
正如你所看到的，這些通道涵蓋了UDP 和 TCP 網絡IO，以及文件IO。
與這些類一起的有一些有趣的接口，但為簡單起見，我盡量在概述中不提到它們。本教程其它章節與它們相關的地方我會進行解釋。
以下是Java NIO里關鍵的Buffer實現：
- ByteBuffer
- CharBuffer
- DoubleBuffer
- FloatBuffer
- IntBuffer
- LongBuffer
- ShortBuffer
這些Buffer覆蓋了你能通過IO發送的基本數據類型：byte, short, int, long, float, double 和 char。
Java NIO 還有個 MappedByteBuffer，用於表示內存映射文件， 我也不打算在概述中說明。
Selector
Selector允許單線程處理多個 Channel。如果你的應用打開了多個連接（通道），但每個連接的流量都很低，使用Selector就會很方便。例如，在一個聊天服務器中。
這是在一個單線程中使用一個Selector處理3個Channel的圖示：
要使用Selector，得向Selector注冊Channel，然后調用它的select()方法。這個方法會一直阻塞到某個注冊的通道有事件就緒。一旦這個方法返回，線程就可以處理這些事件，事件的例子有如新連接進來，數據接收等。
NIO使用步驟
服務端步驟：
步驟一：打開ServerSocketChannel，用於監聽客戶端的連接，它是所有客戶端連接的父管道，代碼示例如下：
ServerSocketChannel acceptorSvr = ServerSocketChannel.open();
步驟二：綁定監聽端口，設置連接為非阻塞模式，示例代碼如下：
acceptorSvr.socket().bind(new InetSocketAddress(InetAddress.getByName(“IP”), port)); acceptorSvr.configureBlocking(false);
步驟三：創建Reactor線程，創建多路復用器並啟動線程，代碼如下：
Selector selector = Selector.open(); New Thread(new ReactorTask()).start();
步驟四：將ServerSocketChannel注冊到Reactor線程的多路復用器Selector上，監聽ACCEPT事件，代碼如下：
SelectionKey key = acceptorSvr.register( selector, SelectionKey.OP_ACCEPT, ioHandler);
步驟五：多路復用器在線程run方法的無限循環體內輪詢准備就緒的Key，代碼如下：
int num = selector.select(); Set selectedKeys = selector.selectedKeys(); Iterator it = selectedKeys.iterator(); while (it.hasNext()) { SelectionKey key = (SelectionKey)it.next(); // ... deal with I/O event ... }
步驟六：多路復用器監聽到有新的客戶端接入，處理新的接入請求，完成TCP三次握手，建立物理鏈路，代碼示例如下：
SocketChannel channel = svrChannel.accept();
步驟七：設置客戶端鏈路為非阻塞模式，示例代碼如下：
channel.configureBlocking(false); channel.socket().setReuseAddress(true);
步驟八：將新接入的客戶端連接注冊到Reactor線程的多路復用器上，監聽讀操作，用來讀取客戶端發送的網絡消息，代碼如下：
SelectionKey key = socketChannel.register( selector, SelectionKey.OP_READ, ioHandler);
步驟九：異步讀取客戶端請求消息到緩沖區，示例代碼如下：
int readNumber = channel.read(receivedBuffer);
步驟十：對ByteBuffer進行編解碼，如果有半包消息指針reset，繼續讀取后續的報文，將解碼成功的消息封裝成Task，投遞到業務線程池中，進行業務邏輯編排，示例代碼如下：
Object message = null; while(buffer.hasRemain()) { byteBuffer.mark(); Object message = decode(byteBuffer); if (message == null) { byteBuffer.reset(); break; } messageList.add(message ); } if (!byteBuffer.hasRemain()) byteBuffer.clear(); else byteBuffer.compact(); if (messageList != null & !messageList.isEmpty()) { for(Object messageE : messageList) handlerTask(messageE); }
步驟十一：將POJO對象encode成ByteBuffer，調用SocketChannel的異步write接口，將消息異步發送給客戶端，示例代碼如下：
socketChannel.write(buffer);
注意：如果發送區TCP緩沖區滿，會導致寫半包，此時，需要注冊監聽寫操作位，循環寫，直到整包消息寫入TCP緩沖區，此處不贅述，后續Netty源碼分析章節會詳細分析Netty的處理策略。
當我們了解創建NIO服務端的基本步驟之后，下面我們將前面的時間服務器程序通過NIO重寫一遍，讓大家能夠學習到完整版的NIO服務端創建。
客戶端步驟：
步驟一：打開SocketChannel，綁定客戶端本地地址（可選，默認系統會隨機分配一個可用的本地地址），示例代碼如下：
SocketChannel clientChannel = SocketChannel.open();
步驟二：設置SocketChannel為非阻塞模式，同時設置客戶端連接的TCP參數，示例代碼如下：
clientChannel.configureBlocking(false); socket.setReuseAddress(true); socket.setReceiveBufferSize(BUFFER_SIZE); socket.setSendBufferSize(BUFFER_SIZE);
步驟三：異步連接服務端，示例代碼如下：
boolean connected = clientChannel.connect(new InetSocketAddress(“ip”,port));
步驟四：判斷是否連接成功，如果連接成功，則直接注冊讀狀態位到多路復用器中，如果當前沒有連接成功（異步連接，返回false，說明客戶端已經發送sync包，服務端沒有返回ack包，物理鏈路還沒有建立），示例代碼如下：
if (connected) { clientChannel.register( selector, SelectionKey.OP_READ, ioHandler); } else { clientChannel.register( selector, SelectionKey.OP_CONNECT, ioHandler); }
步驟五：向Reactor線程的多路復用器注冊OP_CONNECT狀態位，監聽服務端的TCP ACK應答，示例代碼如下：
clientChannel.register( selector, SelectionKey.OP_CONNECT, ioHandler);
步驟六：創建Reactor線程，創建多路復用器並啟動線程，代碼如下：
Selector selector = Selector.open(); New Thread(new ReactorTask()).start();
步驟七：多路復用器在線程run方法的無限循環體內輪詢准備就緒的Key，代碼如下：
int num = selector.select(); Set selectedKeys = selector.selectedKeys(); Iterator it = selectedKeys.iterator(); while (it.hasNext()) { if (key.isConnectable()) //handlerConnect(); }
步驟九：判斷連接結果，如果連接成功，注冊讀事件到多路復用器，示例代碼如下：
if (channel.finishConnect()) registerRead();
步驟十：注冊讀事件到多路復用器：
clientChannel.register( selector, SelectionKey.OP_READ, ioHandler);
步驟十一：異步讀客戶端請求消息到緩沖區，示例代碼如下：
int readNumber = channel.read(receivedBuffer);
步驟十二：對ByteBuffer進行編解碼，如果有半包消息接收緩沖區Reset，繼續讀取后續的報文，將解碼成功的消息封裝成Task，投遞到業務線程池中，進行業務邏輯編排，示例代碼如下：
Object message = null; while(buffer.hasRemain()) { byteBuffer.mark(); Object message = decode(byteBuffer); if (message == null) { byteBuffer.reset(); break; } messageList.add(message ); } if (!byteBuffer.hasRemain()) byteBuffer.clear(); else byteBuffer.compact(); if (messageList != null & !messageList.isEmpty()) { for(Object messageE : messageList) handlerTask(messageE); }
步驟十三：將POJO對象encode成ByteBuffer，調用SocketChannel的異步write接口，將消息異步發送給客戶端，示例代碼如下：
完整代碼：
package com.dxz.springsession.nio.demo6; import java.io.IOException; import java.net.InetAddress; import java.net.InetSocketAddress; import java.net.UnknownHostException; 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; public class NIOServer { public static void main(String[] args) throws IOException { new Thread(new ReactorTask()).start(); } public static class ReactorTask implements Runnable { private Selector selector; public ReactorTask() { try { // 第一步：打開ServerSocketChannel，用於監聽客戶端的連接，它是所有客戶端連接的父管道 ServerSocketChannel acceptorSvr = ServerSocketChannel.open(); // 第二步：病毒監聽端口，設置連接為非阻塞模式 acceptorSvr.socket().bind(new InetSocketAddress(InetAddress.getByName("localhost"), 1234)); acceptorSvr.configureBlocking(false); // 第三步：創建Reactor線程，創建多路復用器並啟動線程 selector = Selector.open(); // 第四步：將ServerSocketChannel注冊到Reactor線程的多路復用器Selector上，監聽Accept事件 SelectionKey key = acceptorSvr.register(selector, SelectionKey.OP_ACCEPT); } catch (UnknownHostException e) { e.printStackTrace(); } catch (IOException e) { e.printStackTrace(); } } @Override public void run() { // 第五步：在run方法中無限循環體內輪詢准備就緒的Key while (true) { try { selector.select(1000); Set<SelectionKey> selectedKeys = selector.selectedKeys(); Iterator<SelectionKey> it = selectedKeys.iterator(); SelectionKey key = null; while (it.hasNext()) { key = it.next(); it.remove(); try { if (key.isValid()) { // 處理新接入的請求消息 if (key.isAcceptable()) { // 第六步：多路復用器監聽到有新的客戶端接入，處理新的接入請求，完成TCP三次握手，建立物理鏈路 ServerSocketChannel ssc = (ServerSocketChannel) key.channel(); SocketChannel sc = ssc.accept(); // 第七步：設置客戶端鏈路為非阻塞模式 sc.configureBlocking(false); sc.socket().setReuseAddress(true); // 第八步：將新接入的客戶端連接注冊到Reactor線程的多路復用器上，監聽讀操作，讀取客戶端發送的網絡消息 sc.register(selector, SelectionKey.OP_READ); } if (key.isReadable()) { // 第九步：異步讀取客戶端請求消息到緩存區 SocketChannel sc = (SocketChannel) key.channel(); ByteBuffer readBuffer = ByteBuffer.allocate(1024); int readBytes = sc.read(readBuffer); // 第十步：對ByteBuffer進行編解碼，如果有半包消息指針reset，繼續讀取后續的報文 if (readBytes > 0) { readBuffer.flip(); byte[] bytes = new byte[readBuffer.remaining()]; readBuffer.get(bytes); String body = new String(bytes, "UTF-8"); System.out.println("The time server receive order : " + body); String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body) ? new java.util.Date(System.currentTimeMillis()).toString() : "BAD ORDER"; //寫應答 byte[] bytes2 = currentTime.getBytes(); ByteBuffer writeBuffer = ByteBuffer.allocate(bytes2.length); writeBuffer.put(bytes2); writeBuffer.flip(); sc.write(writeBuffer); } else if (readBytes < 0) { // 對端鏈路關閉 key.cancel(); sc.close(); } else ; // 讀到0字節，忽略 } } } catch (Exception e) { if (key != null) { key.cancel(); if (key.channel() != null) key.channel().close(); } } } } catch (Throwable t) { t.printStackTrace(); } } } } }
客戶端：
package com.dxz.springsession.nio.demo6; 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; public class TimeClientHandle implements Runnable { private String host; private int port; private Selector selector; private SocketChannel socketChannel; private volatile boolean stop; public TimeClientHandle(String host, int port) { this.host = host == null ? "127.0.0.1" : host; this.port = port; try { //第一步：打開SocketChannel,用於創建客戶端連接 socketChannel = SocketChannel.open(); //第二步：設置SocketChannel為非阻塞模式 socketChannel.configureBlocking(false); //第三步：創建多路復用器（在Reactor線程中） selector = Selector.open(); } catch (IOException e) { e.printStackTrace(); System.exit(1); } } @Override public void run() { try { // 第四步：socketChannel發起連接 if (socketChannel.connect(new InetSocketAddress(host, port))) { //第五步：如果直接連接成功，則注冊到多路復用器上 socketChannel.register(selector, SelectionKey.OP_READ); //第六步：發送請求消息，讀應答 byte[] req = "QUERY TIME ORDER".getBytes(); ByteBuffer writeBuffer = ByteBuffer.allocate(req.length); writeBuffer.put(req); writeBuffer.flip(); socketChannel.write(writeBuffer); if (!writeBuffer.hasRemaining()) System.out.println("Send order 2 server succeed."); } else socketChannel.register(selector, SelectionKey.OP_CONNECT); } catch (IOException e) { e.printStackTrace(); System.exit(1); } while (!stop) { try { //第七步：多路復用器在run的無限循環體內輪詢准備就緒的Key selector.select(1000); Set<SelectionKey> selectedKeys = selector.selectedKeys(); Iterator<SelectionKey> it = selectedKeys.iterator(); SelectionKey key = null; while (it.hasNext()) { key = it.next(); it.remove(); try { if (key.isValid()) { //第八步：將連接成功的Channel注冊到多路復用器上 // 判斷是否連接成功 SocketChannel sc = (SocketChannel) key.channel(); if (key.isConnectable()) { if (sc.finishConnect()) { sc.register(selector, SelectionKey.OP_READ); //發送請求消息，讀應答 byte[] req = "QUERY TIME ORDER".getBytes(); ByteBuffer writeBuffer = ByteBuffer.allocate(req.length); writeBuffer.put(req); writeBuffer.flip(); sc.write(writeBuffer); if (!writeBuffer.hasRemaining()) System.out.println("Send order 2 server succeed."); } else System.exit(1);// 連接失敗，進程退出 } //監聽讀操作，讀取服務端寫回的網絡信息 if (key.isReadable()) { //第九步：讀取信息到緩沖區 ByteBuffer readBuffer = ByteBuffer.allocate(1024); int readBytes = sc.read(readBuffer); if (readBytes > 0) { readBuffer.flip(); byte[] bytes = new byte[readBuffer.remaining()]; readBuffer.get(bytes); String body = new String(bytes, "UTF-8"); System.out.println("Now is : " + body); this.stop = true; } else if (readBytes < 0) { // 對端鏈路關閉 key.cancel(); sc.close(); } else ; // 讀到0字節，忽略 } } } catch (Exception e) { if (key != null) { key.cancel(); if (key.channel() != null) key.channel().close(); } } } } catch (Exception e) { e.printStackTrace(); System.exit(1); } } // 多路復用器關閉后，所有注冊在上面的Channel和Pipe等資源都會被自動去注冊並關閉，所以不需要重復釋放資源 if (selector != null) try { selector.close(); } catch (IOException e) { e.printStackTrace(); } } } package com.dxz.springsession.nio.demo6; public class TimeClient { /** * @param args */ public static void main(String[] args) { int port = 1234; if (args != null && args.length > 0) { try { port = Integer.valueOf(args[0]); } catch (NumberFormatException e) { // 采用默認值 } } new Thread(new TimeClientHandle("127.0.0.1", port), "TimeClient-001") .start(); } }