Java NIO原理 圖文分析及代碼實(shí)現Java NIO原理圖文分析及代碼實(shí)現
前言:
最近在分析hadoop的RPC(Remote Procedure Call Protocol ，遠程過(guò)程調用協(xié)議，它是一種通過(guò)網(wǎng)絡(luò )從遠程計算機程序上請求服務(wù)，而不需要了解底層網(wǎng)絡(luò )技術(shù)的協(xié)議?？梢詤⒖迹篽ttp://baike.baidu.com/view/32726.htm)機制時(shí)，發(fā)現hadoop的RPC機制的實(shí)現主要用到了兩個(gè)技術(shù)：動(dòng)態(tài)代理（動(dòng)態(tài)代理可以參考博客：http://weixiaolu.iteye.com/blog/1477774）和java NIO。為了能夠正確地分析hadoop的RPC源碼，我覺(jué)得很有必要先研究一下java NIO的原理和具體實(shí)現。
這篇博客我主要從兩個(gè)方向來(lái)分析java NIO
目錄：
一．java NIO 和阻塞I/O的區別
1. 阻塞I/O通信模型
2. java NIO原理及通信模型
二．java NIO服務(wù)端和客戶(hù)端代碼實(shí)現
具體分析：
一．java NIO 和阻塞I/O的區別
1. 阻塞I/O通信模型
假如現在你對阻塞I/O已有了一定了解，我們知道阻塞I/O在調用InputStream.read()方法時(shí)是阻塞的，它會(huì )一直等到數據到來(lái)時(shí)（或超時(shí)）才會(huì )返回；同樣，在調用ServerSocket.accept()方法時(shí)，也會(huì )一直阻塞到有客戶(hù)端連接才會(huì )返回，每個(gè)客戶(hù)端連接過(guò)來(lái)后，服務(wù)端都會(huì )啟動(dòng)一個(gè)線(xiàn)程去處理該客戶(hù)端的請求。阻塞I/O的通信模型示意圖如下：
如果你細細分析，一定會(huì )發(fā)現阻塞I/O存在一些缺點(diǎn)。根據阻塞I/O通信模型，我總結了它的兩點(diǎn)缺點(diǎn)：
1. 當客戶(hù)端多時(shí)，會(huì )創(chuàng )建大量的處理線(xiàn)程。且每個(gè)線(xiàn)程都要占用?？臻g和一些CPU時(shí)間
2. 阻塞可能帶來(lái)平凡的上下文切換，且大部分上下文切換可能是無(wú)意義的。
在這種情況下非阻塞式I/O就有了它的應用前景。
2. java NIO原理及通信模型
Java NIO是在jdk1.4開(kāi)始使用的，它既可以說(shuō)成“新I/O”，也可以說(shuō)成非阻塞式I/O。下面是java NIO的工作原理：
1. 由一個(gè)專(zhuān)門(mén)的線(xiàn)程來(lái)處理所有的 IO 事件，并負責分發(fā)。
2. 事件驅動(dòng)機制：事件到的時(shí)候觸發(fā)，而不是同步的去監視事件。
3. 線(xiàn)程通訊：線(xiàn)程之間通過(guò) wait,notify 等方式通訊。保證每次上下文切換都是有意義的。減少無(wú)謂的進(jìn)程切換。
閱讀過(guò)一些資料之后，下面貼出我理解的java NIO的工作原理圖：
（注：每個(gè)線(xiàn)程的處理流程大概都是讀取數據、解碼、計算處理、編碼、發(fā)送響應。）
Java NIO的服務(wù)端只需啟動(dòng)一個(gè)專(zhuān)門(mén)的線(xiàn)程來(lái)處理所有的 IO 事件，這種通信模型是怎么實(shí)現的呢？呵呵，我們一起來(lái)探究它的奧秘吧。java NIO采用了雙向通道（channel）進(jìn)行數據傳輸，而不是單向的流（stream），在通道上可以注冊我們感興趣的事件。一共有以下四種事件：
事件名對應值
服務(wù)端接收客戶(hù)端連接事件SelectionKey.OP_ACCEPT(16)
客戶(hù)端連接服務(wù)端事件SelectionKey.OP_CONNECT(8)
讀事件SelectionKey.OP_READ(1)
寫(xiě)事件SelectionKey.OP_WRITE(4)
服務(wù)端和客戶(hù)端各自維護一個(gè)管理通道的對象，我們稱(chēng)之為selector，該對象能檢測一個(gè)或多個(gè)通道 (channel) 上的事件。我們以服務(wù)端為例，如果服務(wù)端的通道上注冊了讀事件，某時(shí)刻客戶(hù)端給服務(wù)端發(fā)送了一些數據，阻塞I/O這時(shí)會(huì )調用read()方法阻塞地讀取數據，而NIO的服務(wù)端會(huì )在selector中添加一個(gè)讀事件。服務(wù)端的處理線(xiàn)程會(huì )輪詢(xún)地訪(fǎng)問(wèn)selector，如果訪(fǎng)問(wèn)selector時(shí)發(fā)現有感興趣的事件到達，則處理這些事件，如果沒(méi)有感興趣的事件到達，則處理線(xiàn)程會(huì )一直阻塞直到感興趣的事件到達為止。下面是我理解的java NIO的通信模型示意圖：
二．java NIO服務(wù)端和客戶(hù)端代碼實(shí)現
為了更好地理解java NIO,下面貼出服務(wù)端和客戶(hù)端的簡(jiǎn)單代碼實(shí)現。
服務(wù)端：
Java代碼
package cn.nio;
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.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
/**
* NIO服務(wù)端
* @author 小路
*/
public class NIOServer {
//通道管理器
private Selector selector;
/**
* 獲得一個(gè)ServerSocket通道，并對該通道做一些初始化的工作
* @param port 綁定的端口號
* @throws IOException
*/
public void initServer(int port) throws IOException {
// 獲得一個(gè)ServerSocket通道
ServerSocketChannel serverChannel = ServerSocketChannel.open();
// 設置通道為非阻塞
serverChannel.configureBlocking(false);
// 將該通道對應的ServerSocket綁定到port端口
serverChannel.socket().bind(new InetSocketAddress(port));
// 獲得一個(gè)通道管理器
this.selector = Selector.open();
//將通道管理器和該通道綁定，并為該通道注冊SelectionKey.OP_ACCEPT事件,注冊該事件后，
//當該事件到達時(shí)，selector.select()會(huì )返回，如果該事件沒(méi)到達selector.select()會(huì )一直阻塞。
serverChannel.register(selector, SelectionKey.OP_ACCEPT);
}
/**
* 采用輪詢(xún)的方式監聽(tīng)selector上是否有需要處理的事件，如果有，則進(jìn)行處理
* @throws IOException
*/
@SuppressWarnings("unchecked")
public void listen() throws IOException {
System.out.println("服務(wù)端啟動(dòng)成功！");
// 輪詢(xún)訪(fǎng)問(wèn)selector
while (true) {
//當注冊的事件到達時(shí)，方法返回；否則,該方法會(huì )一直阻塞
selector.select();
// 獲得selector中選中的項的迭代器，選中的項為注冊的事件
Iterator ite = this.selector.selectedKeys().iterator();
while (ite.hasNext()) {
SelectionKey key = (SelectionKey) ite.next();
// 刪除已選的key,以防重復處理
ite.remove();
// 客戶(hù)端請求連接事件
if (key.isAcceptable()) {
ServerSocketChannel server = (ServerSocketChannel) key
.channel();
// 獲得和客戶(hù)端連接的通道
SocketChannel channel = server.accept();
// 設置成非阻塞
channel.configureBlocking(false);
//在這里可以給客戶(hù)端發(fā)送信息哦
channel.write(ByteBuffer.wrap(new String("向客戶(hù)端發(fā)送了一條信息").getBytes()));
//在和客戶(hù)端連接成功之后，為了可以接收到客戶(hù)端的信息，需要給通道設置讀的權限。
channel.register(this.selector, SelectionKey.OP_READ);
// 獲得了可讀的事件
} else if (key.isReadable()) {
read(key);
}
}
}
}
/**
* 處理讀取客戶(hù)端發(fā)來(lái)的信息 的事件
* @param key
* @throws IOException
*/
public void read(SelectionKey key) throws IOException{
// 服務(wù)器可讀取消息:得到事件發(fā)生的Socket通道
SocketChannel channel = (SocketChannel) key.channel();
// 創(chuàng )建讀取的緩沖區
ByteBuffer buffer = ByteBuffer.allocate(10);
channel.read(buffer);
byte[] data = buffer.array();
String msg = new String(data).trim();
System.out.println("服務(wù)端收到信息："+msg);
ByteBuffer outBuffer = ByteBuffer.wrap(msg.getBytes());
channel.write(outBuffer);// 將消息回送給客戶(hù)端
}
/**
* 啟動(dòng)服務(wù)端測試
* @throws IOException
*/
public static void main(String[] args) throws IOException {
NIOServer server = new NIOServer();
server.initServer(8000);
server.listen();
}
}
客戶(hù)端：
Java代碼
/**
*
*/
package cn.nio;
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;
/**
* NIO客戶(hù)端
* @author 小路
*/
public class NIOClient {
//通道管理器
private Selector selector;
/**
* 獲得一個(gè)Socket通道，并對該通道做一些初始化的工作
* @param ip 連接的服務(wù)器的ip
* @param port 連接的服務(wù)器的端口號
* @throws IOException
*/
public void initClient(String ip,int port) throws IOException {
// 獲得一個(gè)Socket通道
SocketChannel channel = SocketChannel.open();
// 設置通道為非阻塞
channel.configureBlocking(false);
// 獲得一個(gè)通道管理器
this.selector = Selector.open();
// 客戶(hù)端連接服務(wù)器,其實(shí)方法執行并沒(méi)有實(shí)現連接，需要在listen（）方法中調
//用channel.finishConnect();才能完成連接
channel.connect(new InetSocketAddress(ip,port));
//將通道管理器和該通道綁定，并為該通道注冊SelectionKey.OP_CONNECT事件,注冊該事件后，
//當該事件到達時(shí)，selector.select()會(huì )返回，如果該事件沒(méi)到達selector.select()會(huì )一直阻塞。
channel.register(selector, SelectionKey.OP_CONNECT);
}
/**
* 采用輪詢(xún)的方式監聽(tīng)selector上是否有需要處理的事件，如果有，則進(jìn)行處理
* @throws IOException
*/
@SuppressWarnings("unchecked")
public void listen() throws IOException {
// 輪詢(xún)訪(fǎng)問(wèn)selector
while (true) {
// 選擇一組可以進(jìn)行I/O操作的事件，放在selector中,該方法會(huì )阻塞
selector.select();
// 獲得selector中選中的項的迭代器
Iterator ite = this.selector.selectedKeys().iterator();
while (ite.hasNext()) {
SelectionKey key = (SelectionKey) ite.next();
// 刪除已選的key,以防重復處理
ite.remove();
// 連接事件發(fā)生
if (key.isConnectable()) {
SocketChannel channel = (SocketChannel) key
.channel();
// 如果正在連接，則完成連接
if(channel.isConnectionPending()){
channel.finishConnect();
}
// 設置成非阻塞
channel.configureBlocking(false);
//在這里可以給服務(wù)端發(fā)送信息哦
channel.write(ByteBuffer.wrap(new String("向服務(wù)端發(fā)送了一條信息").getBytes()));
//在和服務(wù)端連接成功之后，為了可以接收到服務(wù)端的信息，需要給通道設置讀的權限。
channel.register(this.selector, SelectionKey.OP_READ);
// 獲得了可讀的事件
} else if (key.isReadable()) {
read(key);
}
}
}
}
/**
* 處理讀取服務(wù)端發(fā)來(lái)的信息 的事件
* @param key
* @throws IOException
*/
public void read(SelectionKey key) throws IOException{
//和服務(wù)端的read方法一樣
}
/**
* 啟動(dòng)客戶(hù)端測試
* @throws IOException
*/
public static void main(String[] args) throws IOException {
NIOClient client = new NIOClient();
client.initClient("localhost",8000);
client.listen();
}
}
小結：
終于把動(dòng)態(tài)代理和java NIO分析完了，呵呵，下面就要分析hadoop的RPC機制源碼了。不過(guò)如果對java NIO的理解存在異議的，歡迎一起討論。
