前言:
最近在分析hadoop的RPC(Remote Procedure Call Protocol ，遠程過程調(diào)用協(xié)議，它是一種通過網(wǎng)絡(luò)從遠程計算機程序上請求服務(wù)，而不需要了解底層網(wǎng)絡(luò)技術(shù)的協(xié)議。可以參考：http://baike.baidu.com/view/32726.htm ）機制時，發(fā)現(xiàn)hadoop的RPC機制的實現(xiàn)主要用到了兩個技術(shù)：動態(tài)代理（動態(tài)代理可以參考博客：http://weixiaolu.iteye.com/blog/1477774 ）和java NIO。為了能夠正確地分析hadoop的RPC源碼，我覺得很有必要先研究一下java NIO的原理和具體實現(xiàn)。
這篇博客我主要從兩個方向來分析java NIO
目錄：
一．java NIO 和阻塞I/O的區(qū)別
1. 阻塞I/O通信模型
2. java NIO原理及通信模型
二．java NIO服務(wù)端和客戶端代碼實現(xiàn)
具體分析：
一．java NIO 和阻塞I/O的區(qū)別
1. 阻塞I/O通信模型
假如現(xiàn)在你對阻塞I/O已有了一定了解，我們知道阻塞I/O在調(diào)用InputStream.read()方法時是阻塞的，它會一直等到數(shù)據(jù)到來時（或超時）才會返回；同樣，在調(diào)用ServerSocket.accept()方法時，也會一直阻塞到有客戶端連接才會返回，每個客戶端連接過來后，服務(wù)端都會啟動一個線程去處理該客戶端的請求。阻塞I/O的通信模型示意圖如下：
如果你細細分析，一定會發(fā)現(xiàn)阻塞I/O存在一些缺點。根據(jù)阻塞I/O通信模型，我總結(jié)了它的兩點缺點：
1. 當(dāng)客戶端多時，會創(chuàng)建大量的處理線程。且每個線程都要占用棧空間和一些CPU時間
2. 阻塞可能帶來頻繁的上下文切換，且大部分上下文切換可能是無意義的。
在這種情況下非阻塞式I/O就有了它的應(yīng)用前景。
2. java NIO原理及通信模型
Java NIO是在jdk1.4開始使用的，它既可以說成“新I/O”，也可以說成非阻塞式I/O。下面是java NIO的工作原理：
1. 由一個專門的線程來處理所有的 IO 事件，并負責(zé)分發(fā)。
2. 事件驅(qū)動機制：事件到的時候觸發(fā)，而不是同步的去監(jiān)視事件。
3. 線程通訊：線程之間通過 wait,notify 等方式通訊。保證每次上下文切換都是有意義的。減少無謂的線程切換。
閱讀過一些資料之后，下面貼出我理解的java NIO的工作原理圖：
（注：每個線程的處理流程大概都是讀取數(shù)據(jù)、解碼、計算處理、編碼、發(fā)送響應(yīng)。）
Java NIO的服務(wù)端只需啟動一個專門的線程來處理所有的 IO 事件，這種通信模型是怎么實現(xiàn)的呢？呵呵，我們一起來探究它的奧秘吧。java NIO采用了雙向通道（channel）進行數(shù)據(jù)傳輸，而不是單向的流（stream），在通道上可以注冊我們感興趣的事件。一共有以下四種事件：
| 事件名 | 對應(yīng)值 |
| 服務(wù)端接收客戶端連接事件 | SelectionKey.OP_ACCEPT(16) |
| 客戶端連接服務(wù)端事件 | SelectionKey.OP_CONNECT(8) |
| 讀事件 | SelectionKey.OP_READ(1) |
| 寫事件 | SelectionKey.OP_WRITE(4) |
服務(wù)端和客戶端各自維護一個管理通道的對象，我們稱之為selector，該對象能檢測一個或多個通道 (channel) 上的事件。我們以服務(wù)端為例，如果服務(wù)端的selector上注冊了讀事件，某時刻客戶端給服務(wù)端發(fā)送了一些數(shù)據(jù)，阻塞I/O這時會調(diào)用read()方法阻塞地讀取數(shù)據(jù)，而NIO的服務(wù)端會在selector中添加一個讀事件。服務(wù)端的處理線程會輪詢地訪問selector，如果訪問selector時發(fā)現(xiàn)有感興趣的事件到達，則處理這些事件，如果沒有感興趣的事件到達，則處理線程會一直阻塞直到感興趣的事件到達為止。下面是我理解的java NIO的通信模型示意圖：
二．java NIO服務(wù)端和客戶端代碼實現(xiàn)
為了更好地理解java NIO,下面貼出服務(wù)端和客戶端的簡單代碼實現(xiàn)。
服務(wù)端：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
|
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; /** * 獲得一個ServerSocket通道，并對該通道做一些初始化的工作 * @param port 綁定的端口號 * @throws IOException */ public void initServer(int port) throws IOException { // 獲得一個ServerSocket通道 ServerSocketChannel serverChannel = ServerSocketChannel.open(); // 設(shè)置通道為非阻塞 serverChannel.configureBlocking(false); // 將該通道對應(yīng)的ServerSocket綁定到port端口 serverChannel.socket().bind(new InetSocketAddress(port)); // 獲得一個通道管理器 this.selector = Selector.open(); //將通道管理器和該通道綁定，并為該通道注冊SelectionKey.OP_ACCEPT事件,注冊該事件后， //當(dāng)該事件到達時，selector.select()會返回，如果該事件沒到達selector.select()會一直阻塞。 serverChannel.register(selector, SelectionKey.OP_ACCEPT); } /** * 采用輪詢的方式監(jiān)聽selector上是否有需要處理的事件，如果有，則進行處理 * @throws IOException */ @SuppressWarnings("unchecked") public void listen() throws IOException { System.out.println("服務(wù)端啟動成功！"); // 輪詢訪問selector while (true) { //當(dāng)注冊的事件到達時，方法返回；否則,該方法會一直阻塞 selector.select(); // 獲得selector中選中的項的迭代器，選中的項為注冊的事件 Iterator ite = this.selector.selectedKeys().iterator(); while (ite.hasNext()) { SelectionKey key = (SelectionKey) ite.next(); // 刪除已選的key,以防重復(fù)處理 ite.remove(); // 客戶端請求連接事件 if (key.isAcceptable()) { ServerSocketChannel server = (ServerSocketChannel) key .channel(); // 獲得和客戶端連接的通道 SocketChannel channel = server.accept(); // 設(shè)置成非阻塞 channel.configureBlocking(false); //在這里可以給客戶端發(fā)送信息哦 channel.write(ByteBuffer.wrap(new String("向客戶端發(fā)送了一條信息").getBytes())); //在和客戶端連接成功之后，為了可以接收到客戶端的信息，需要給通道設(shè)置讀的權(quán)限。 channel.register(this.selector, SelectionKey.OP_READ); // 獲得了可讀的事件 } else if (key.isReadable()) { read(key); } } } } /** * 處理讀取客戶端發(fā)來的信息 的事件 * @param key * @throws IOException */ public void read(SelectionKey key) throws IOException{ // 服務(wù)器可讀取消息:得到事件發(fā)生的Socket通道 SocketChannel channel = (SocketChannel) key.channel(); // 創(chuàng)建讀取的緩沖區(qū) 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);// 將消息回送給客戶端 } /** * 啟動服務(wù)端測試 * @throws IOException */ public static void main(String[] args) throws IOException { NIOServer server = new NIOServer(); server.initServer(8000); server.listen(); } } |
客戶端：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
|
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客戶端 * @author 小路 */public class NIOClient { //通道管理器 private Selector selector; /** * 獲得一個Socket通道，并對該通道做一些初始化的工作 * @param ip 連接的服務(wù)器的ip * @param port 連接的服務(wù)器的端口號 * @throws IOException */ public void initClient(String ip,int port) throws IOException { // 獲得一個Socket通道 SocketChannel channel = SocketChannel.open(); // 設(shè)置通道為非阻塞 channel.configureBlocking(false); // 獲得一個通道管理器 this.selector = Selector.open(); // 客戶端連接服務(wù)器,其實方法執(zhí)行并沒有實現(xiàn)連接，需要在listen（）方法中調(diào) //用channel.finishConnect();才能完成連接 channel.connect(new InetSocketAddress(ip,port)); //將通道管理器和該通道綁定，并為該通道注冊SelectionKey.OP_CONNECT事件。 channel.register(selector, SelectionKey.OP_CONNECT); } /** * 采用輪詢的方式監(jiān)聽selector上是否有需要處理的事件，如果有，則進行處理 * @throws IOException */ @SuppressWarnings("unchecked") public void listen() throws IOException { // 輪詢訪問selector while (true) { selector.select(); // 獲得selector中選中的項的迭代器 Iterator ite = this.selector.selectedKeys().iterator(); while (ite.hasNext()) { SelectionKey key = (SelectionKey) ite.next(); // 刪除已選的key,以防重復(fù)處理 ite.remove(); // 連接事件發(fā)生 if (key.isConnectable()) { SocketChannel channel = (SocketChannel) key .channel(); // 如果正在連接，則完成連接 if(channel.isConnectionPending()){ channel.finishConnect(); } // 設(shè)置成非阻塞 channel.configureBlocking(false); //在這里可以給服務(wù)端發(fā)送信息哦 channel.write(ByteBuffer.wrap(new String("向服務(wù)端發(fā)送了一條信息").getBytes())); //在和服務(wù)端連接成功之后，為了可以接收到服務(wù)端的信息，需要給通道設(shè)置讀的權(quán)限。 channel.register(this.selector, SelectionKey.OP_READ); // 獲得了可讀的事件 } else if (key.isReadable()) { read(key); } } } } /** * 處理讀取服務(wù)端發(fā)來的信息 的事件 * @param key * @throws IOException */ public void read(SelectionKey key) throws IOException{ //和服務(wù)端的read方法一樣 } /** * 啟動客戶端測試 * @throws IOException */ public static void main(String[] args) throws IOException { NIOClient client = new NIOClient(); client.initClient("localhost",8000); client.listen(); } } |
小結(jié)：
終于把動態(tài)代理和java NIO分析完了，呵呵，下面就要分析hadoop的RPC機制源碼了，博客地址：http://weixiaolu.iteye.com/blog/1504898 。不過如果對java NIO的理解存在異議的，歡迎一起討論。
如需轉(zhuǎn)載，請注明出處：http://weixiaolu.iteye.com/blog/1479656
