1、reactor（反應器）模式
使用單線程模擬多線程，提高資源利用率和程序的效率，增加系統吞吐量。下面例子比較形象的說明了什么是反應器模式：
一個老板經營一個飯店，
傳統模式 - 來一個客人安排一個服務員招呼，客人很滿意；（相當於一個連接一個線程）
后來客人越來越多，需要的服務員越來越多，資源條件不足以再請更多的服務員了，傳統模式已經不能滿足需求。老板之所以為老板自然有過人之處，老板發現，服務員在為客人服務時，當客人點菜的時候，服務員基本處於等待狀態，（阻塞線程，不做事）。
於是乎就讓服務員在客人點菜的時候，去為其他客人服務，當客人菜點好后再招呼服務員即可。 --反應器（reactor）模式誕生了
飯店的生意紅紅火火，幾個服務員就足以支撐大量的客流量，老板用有限的資源賺了更多的money~~~~^_^
2、NIO中的重要概念 通道、緩沖區、選擇器
通道：類似於流，但是可以異步讀寫數據（流只能同步讀寫），通道是雙向的，（流是單向的），通道的數據總是要先讀到一個buffer 或者 從一個buffer寫入，即通道與buffer進行數據交互。
通道類型：
-
- FileChannel：從文件中讀寫數據。
- DatagramChannel：能通過UDP讀寫網絡中的數據。
- SocketChannel：能通過TCP讀寫網絡中的數據。
- ServerSocketChannel：可以監聽新進來的TCP連接，像Web服務器那樣。對每一個新進來的連接都會創建一個SocketChannel。
FileChannel比較特殊，它可以與通道進行數據交互， 不能切換到非阻塞模式，套接字通道可以切換到非阻塞模式；
緩沖區 - 本質上是一塊可以存儲數據的內存，被封裝成了buffer對象而已！
緩沖區類型：
-
- ByteBuffer
- MappedByteBuffer
- CharBuffer
- DoubleBuffer
- FloatBuffer
- IntBuffer
- LongBuffer
- ShortBuffer
常用方法：
- allocate() - 分配一塊緩沖區
- put() - 向緩沖區寫數據
- get() - 向緩沖區讀數據
- filp() - 將緩沖區從寫模式切換到讀模式
- clear() - 從讀模式切換到寫模式，不會清空數據，但后續寫數據會覆蓋原來的數據，即使有部分數據沒有讀，也會被遺忘；
- compact() - 從讀數據切換到寫模式，數據不會被清空，會將所有未讀的數據copy到緩沖區頭部，后續寫數據不會覆蓋，而是在這些數據之后寫數據
- mark() - 對position做出標記，配合reset使用
-
- reset() - 將position置為標記值
緩沖區的一些屬性：
-
- capacity - 緩沖區大小，無論是讀模式還是寫模式，此屬性值不會變；
- position - 寫數據時，position表示當前寫的位置，每寫一個數據，會向下移動一個數據單元，初始為0；最大為capacity - 1
切換到讀模式時，position會被置為0，表示當前讀的位置
-
- limit - 寫模式下，limit 相當於capacity 表示最多可以寫多少數據，切換到讀模式時，limit 等於原先的position，表示最多可以讀多少數據。
選擇器：相當於一個觀察者，用來監聽通道感興趣的事件，一個選擇器可以綁定多個通道；
通道向選擇器注冊時，需要指定感興趣的事件，選擇器支持以下事件：
-
- SelectionKey.OP_CONNECT
- SelectionKey.OP_ACCEPT
- SelectionKey.OP_READ
- SelectionKey.OP_WRITE
如果你對不止一種事件感興趣，那么可以用“位或”操作符將常量連接起來，如下：
int interestSet = SelectionKey.OP_READ | SelectionKey.OP_WRITE;
通道向選擇器注冊時，會返回一個 SelectionKey對象，具有如下屬性
-
- interest集合
- ready集合
- Channel
- Selector
- 附加的對象（可選）
用“位與”操作interest 集合和給定的SelectionKey常量，可以確定某個確定的事件是否在interest 集合中。
int interestSet = selectionKey.interestOps(); boolean isInterestedInAccept = interestSet & SelectionKey.OP_ACCEPT;
boolean isInterestedInConnect = interestSet & SelectionKey.OP_CONNECT; boolean isInterestedInRead = interestSet & SelectionKey.OP_READ; boolean isInterestedInWrite = interestSet & SelectionKey.OP_WRITE;
ready 集合是通道已經准備就緒的操作的集合。在一次選擇(Selection)之后，你會首先訪問這個ready set。Selection將在下一小節進行解釋。可以這樣訪問ready集合：
int readySet = selectionKey.readyOps();
也可以使用以下四個方法獲取已就緒事件，返回值為boolean：
selectionKey.isAcceptable(); selectionKey.isConnectable(); selectionKey.isReadable(); selectionKey.isWritable();
可以將一個對象或者更多信息附着到SelectionKey上，即記錄在附加對象上，方法如下：
selectionKey.attach(theObject);
Object attachedObj = selectionKey.attachment();
可以通過選擇器的select方法獲取是否有就緒的通道；
- int select()
- int select(long timeout)
- int selectNow()
返回值表示上次執行select之后，就緒通道的個數。
可以通過selectedKeySet獲取已就緒的通道。返回值是SelectionKey 的集合，處理完相應的通道之后，需要removed 因為Selector不會自己removed
select阻塞后，可以用wakeup喚醒；執行wakeup時，如果沒有阻塞的select 那么執行完wakeup后下一個執行select就會立即返回。
調用close() 方法關閉selector
下面是一個簡單的實例代碼，幫助理解上面的內容：
package com.pt.nio; import java.net.InetAddress; import java.net.InetSocketAddress; import java.nio.ByteBuffer; import java.nio.CharBuffer; import java.nio.channels.SelectionKey; import java.nio.channels.Selector; import java.nio.channels.ServerSocketChannel; import java.nio.channels.SocketChannel; import java.nio.charset.Charset; import java.nio.charset.CharsetDecoder; import java.util.Iterator; import java.util.Set; public class Reactor implements Runnable { public int id = 100001; public int bufferSize = 2048; @Override public void run() { // TODO Auto-generated method stub init(); } public void init() { try { // 創建通道和選擇器 ServerSocketChannel socketChannel = ServerSocketChannel.open(); Selector selector = Selector.open(); InetSocketAddress inetSocketAddress = new InetSocketAddress( InetAddress.getLocalHost(), 4700); socketChannel.socket().bind(inetSocketAddress); // 設置通道非阻塞 綁定選擇器 socketChannel.configureBlocking(false); socketChannel.register(selector, SelectionKey.OP_ACCEPT).attach( id++); System.out.println("Server started .... port:4700"); listener(selector); } catch (Exception e) { // TODO: handle exception } } public void listener(Selector in_selector) { try { while (true) { Thread.sleep(1*1000); in_selector.select(); // 阻塞 直到有就緒事件為止 Set<SelectionKey> readySelectionKey = in_selector .selectedKeys(); Iterator<SelectionKey> it = readySelectionKey.iterator(); while (it.hasNext()) { SelectionKey selectionKey = it.next(); // 判斷是哪個事件 if (selectionKey.isAcceptable()) {// 客戶請求連接 System.out.println(selectionKey.attachment() + " - 接受請求事件"); // 獲取通道 接受連接, // 設置非阻塞模式（必須），同時需要注冊 讀寫數據的事件，這樣有消息觸發時才能捕獲 ServerSocketChannel serverSocketChannel = (ServerSocketChannel) selectionKey .channel(); serverSocketChannel .accept() .configureBlocking(false) .register( in_selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE).attach(id++); System.out .println(selectionKey.attachment() + " - 已連接"); // 下面這種寫法是有問題的 不應該在serverSocketChannel上面注冊 /* * serverSocketChannel.configureBlocking(false); * serverSocketChannel.register(in_selector, * SelectionKey.OP_READ); * serverSocketChannel.register(in_selector, * SelectionKey.OP_WRITE); */ } if (selectionKey.isReadable()) {// 讀數據 System.out.println(selectionKey.attachment() + " - 讀數據事件"); SocketChannel clientChannel=(SocketChannel)selectionKey.channel(); ByteBuffer receiveBuf = ByteBuffer.allocate(bufferSize); clientChannel.read(receiveBuf); System.out.println(selectionKey.attachment() + " - 讀取數據：" + getString(receiveBuf)); } if (selectionKey.isWritable()) {// 寫數據 System.out.println(selectionKey.attachment() + " - 寫數據事件"); SocketChannel clientChannel = (SocketChannel) selectionKey.channel(); ByteBuffer sendBuf = ByteBuffer.allocate(bufferSize); String sendText = "hello\n"; sendBuf.put(sendText.getBytes()); sendBuf.flip(); //寫完數據后調用此方法 clientChannel.write(sendBuf); } if (selectionKey.isConnectable()) { System.out.println(selectionKey.attachment() + " - 連接事件"); } // 必須removed 否則會繼續存在，下一次循環還會進來, // 注意removed 的位置，針對一個.next() remove一次 it.remove(); } } } catch (Exception e) { // TODO: handle exception System.out.println("Error - " + e.getMessage()); e.printStackTrace(); } } /** * ByteBuffer 轉換 String * @param buffer * @return */ public static String getString(ByteBuffer buffer) { String string = ""; try { for(int i = 0; i<buffer.position();i++){ string += (char)buffer.get(i); } return string; } catch (Exception ex) { ex.printStackTrace(); return ""; } } }
package com.pt.bio; import java.io.*; import java.net.*; public class BioServer implements Runnable { @Override public void run() { // TODO Auto-generated method stub System.out.println("Hello Server!!"); try { ServerSocket server = null; try { server = new ServerSocket(4700); // 創建一個ServerSocket在端口4700監聽客戶請求 } catch (Exception e) { System.out.println("can not listen to:" + e); // 出錯，打印出錯信息 } Socket socket = null; try { socket = server.accept(); // 使用accept()阻塞等待客戶請求，有客戶 // 請求到來則產生一個Socket對象，並繼續執行 } catch (Exception e) { System.out.println("Error." + e); // 出錯，打印出錯信息 } String line; BufferedReader is = new BufferedReader(new InputStreamReader( socket.getInputStream())); // 由Socket對象得到輸入流，並構造相應的BufferedReader對象 // 由Socket對象得到輸出流，並構造PrintWriter對象 // BufferedReader sin = new BufferedReader(new InputStreamReader( // System.in)); // 由系統標准輸入設備構造BufferedReader對象 System.out.println("Client:" + is.readLine()); PrintWriter os = new PrintWriter(socket.getOutputStream()); // 在標准輸出上打印從客戶端讀入的字符串 line = "hello"; // 從標准輸入讀入一字符串 // while (!line.equals("bye")) { // 如果該字符串為 "bye"，則停止循環 os.println(line); // 向客戶端輸出該字符串 os.flush(); // 刷新輸出流，使Client馬上收到該字符串 // System.out.println("Server:" + line); // 在系統標准輸出上打印讀入的字符串 // System.out.println("Client:" + is.readLine()); // 從Client讀入一字符串，並打印到標准輸出上 // line = sin.readLine(); // 從系統標准輸入讀入一字符串 // } // 繼續循環 // os.close(); // 關閉Socket輸出流 is.close(); // 關閉Socket輸入流 socket.close(); // 關閉Socket server.close(); // 關閉ServerSocket } catch (Exception e) { System.out.println("Error." + e); // 出錯，打印出錯信息 } } }
package com.pt; import java.io.BufferedReader; import java.io.InputStreamReader; import java.io.PrintWriter; import java.net.Socket; import org.junit.Test; import com.pt.bio.BioServer; import com.pt.nio.Reactor; public class TestReactor { @Test public void testConnect() throws Exception{ Socket socket=new Socket("192.168.82.35",4700);//BIO 阻塞 System.out.println("連接成功"); BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(socket.getInputStream())); //下面這種寫法，不用關閉客戶端，服務器端也是可以收到的 { PrintWriter printWriter = new PrintWriter(socket.getOutputStream(), true); printWriter.println("hi"); printWriter.flush(); } //這種寫法必須關閉客戶端，服務器端才可以收到 NIO不用 { // socket.getOutputStream().write(new byte[]{'h','i'}); // socket.getOutputStream().flush(); //必須關閉BIO服務器才能收到消息.NIO服務器不需要關閉 //socket.close(); } byte[] buf = new byte[2048]; System.out.println("准備讀取數據~~"); while(true){ try { //兩種讀取數據方式 int count = socket.getInputStream().read(buf); //會阻塞 //String readFromServer = bufferedReader.readLine();//可以讀取到數據 會阻塞,直到遇見\n //System.out.println("方式二： 讀取數據" + readFromServer); System.out.println("方式一： 讀取數據" + new String(buf) + " count = " + count); Thread.sleep(1*1000); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } //break; } } @Test public void testNioServer(){ Thread server = new Thread(new Reactor()); server.start(); while(true){ try { Thread.sleep(3*1000); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } } } @Test public void testBioServer(){ Thread server = new Thread(new BioServer()); server.start(); while(true){ try { Thread.sleep(3*1000); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } } } }
其中 testNioServer()方法，是啟動NIO服務器端；
testBioServer()方法是啟動BIO服務器端
testConnect()是BIO的一個連接
基於NIO實現的時鍾服務器：http://www.cnblogs.com/tengpan-cn/p/6529628.html
一篇寫的比較詳細的JAVA NIO的文章：http://www.iteye.com/magazines/132-Java-NIO