国产一级a片免费看高清,亚洲熟女中文字幕在线视频,黄三级高清在线播放,免费黄色视频在线看

NIO+reactor 模式的網(wǎng)路服務器設計方案

1、前言

       在前一篇文章中，介紹了基于 BlockingIO +thread-per-connection 的方案，由于該方案為每一個連接分配一個線程，而線程里的大部分操作都是阻塞式的，所以在高并發(fā)的情況下，會導致產(chǎn)生大量的線程，線程間的上下文切換會浪費大量的 CPU 時間，而且每個線程是需要占用堆?？臻g的，所以服務器能分配的線程數(shù)量也是有限的，當客戶端的并發(fā)訪問量達到一定的數(shù)量級后，服務器的資源就會耗盡，可伸縮性差。

根據(jù)上面的分析，要提高網(wǎng)絡服務器的可伸縮性，就必須解決兩個問題：

• 服務端的線程數(shù)量不能按照客戶端連接數(shù)的增大而成線性增長，但又必須能夠并發(fā)的響應不斷增加的客戶端請求
• 線程里的操作邏輯不能是阻塞式的

因此， java1.4 引入了非阻塞式 NIO （ Non-blocking IO ） , 解決了問題 2 ；而采用基于異步事件通知的 reactor 模式則可以僅僅用一個線程來并發(fā)的為多個連接服務，這樣就解決了問題 1

2、Reactor 模式

2.1 示例

     首先舉一個生活中的例子來比較 thread-per-connection和 reactor方案

某火車票售票廳，只有 1 個售票窗口工作。兩個乘客 a 、 b 先后來購票，由于 a 先到，所以售票窗口先為 a 服務， b 只能排隊

• thread-per-connection ：

    乘客 a與售票窗口開始溝通時，就相當于在客戶端（乘客 a）與服務端（售票廳）之間建立了一個 connection，服務端為每一個 connection分配一個 thread（售票窗口）。當沒有 thread可以分配時，后續(xù)的客戶端請求（乘客 b）就不能及時響應了，所以 b只能排隊。假設存在這種場景，售票窗口的服務員告訴乘客 a票價后，乘客 a準備付款時發(fā)現(xiàn)自己忘記了帶錢包，所以乘客 a打電話給家里人讓他們把錢包送過來，但從 a的家步行到售票廳需要 5分鐘，于是售票窗口的服務員就一直等著（被阻塞），但又不為乘客 b服務，因為她的做事風格（ thread-per-connection）是一定要為一個乘客完完整整服務完后才能接著服務下一位乘客。

   這種情況下，乘客 b 肯定會抱怨，而且 5 分鐘后， b 的后面也肯定排了很多人，售票廳發(fā)現(xiàn)這種情況后，就只能選擇再打開一個售票窗口（分配一個 thread ）為 b 服務，但 b 后面的人也只能排隊。之前那個窗口的服務員一直等著，又不干活，但工資還是照樣拿，所以售票廳（服務端）的開銷很大。

• Reactor

    服務員在等待 a 取錢包的過程中，被通知乘客 b 要求服務，所以窗口和 b 建立連接，悲劇的是 b 也沒有帶錢包，需要家里人送來。此時服務員又被通知 a 的錢包送過來了，所以窗口接著為 a 服務，出票完成后，服務員又被通知 b 的錢包送過來了，所以接著又為 b 服務。這樣，售票廳（服務端）的開銷就小了，現(xiàn)在只需要一個窗口就可以搞定所有事情。

2.2 Reactor 模式的思想：分而治之 + 事件驅(qū)動

• 分而治之：

    一個 connection里發(fā)生的完整的網(wǎng)絡處理過程一般分為 accept、 read、 decode、 compute、 encode、 send這幾步。 Reactor將每個步驟映射為一個 task，服務端端的線程執(zhí)行的最小邏輯單元不再是一次完整的網(wǎng)絡處理過程，而是更小的 task，且采用非阻塞的執(zhí)行方式；

• 事件驅(qū)動：

   每個 task對應一個特定的事件，當 task準備就緒時，對應的事件通知就會發(fā)出。 Reactor收到事件后，分發(fā)給綁定了對應事件的 Handler執(zhí)行 task。

下圖描述了單線程版本的 reactor 模式結(jié)構圖。

關鍵概念：

• Reactor：負責響應事件，分發(fā)給綁定了該事件的 handler執(zhí)行 task
• Handler：綁定了某類事件，負責執(zhí)行該事件對應的 task。
• Acceptor ： Handler 的一種，綁定了 connect 事件。它在系統(tǒng)啟動的時候就已經(jīng)綁定了 connnect 事件，并注冊到 reactor 中。當有客戶端發(fā)起 connect 請求時， reactor 會收到 accept 事件，然后分發(fā)給 acceptor 。 acceptor 首先會 accept 新的連接，然后新建一個 handler ，將其與該連接上得 read 事件綁定，并注冊到 reactor 中。

2.3 基于 reactor 的網(wǎng)絡交互

1. 客戶端連接服務器過程

1)      服務器將綁定了 accept事件的 Acceptor注冊到 Reactor中，準備 accept新的 connection;

2)      服務器啟動 Reactor的事件循環(huán)處理功能（注意：該循環(huán)會阻塞，直到收到事件）

3)      客戶端 connect服務器

4)      Reactor響應 accept事件，分發(fā)給 Acceptor， Acceptor 確定建立一個新的連接。

5)      Acceptor創(chuàng)建一個 handler專門服務該 connection后續(xù)的請求；

6)      Handler綁定該 connection的 read事件，并將自己注冊到 Reactor中

1. 服務器處理客戶端請求過程

1)      客戶端發(fā)送請求

2)      當客戶端的請求數(shù)據(jù)到達服務器時， Reactor響應 read事件，分發(fā)給綁定了 read事件的 handler（即上面第 6步創(chuàng)建的 handler）

3)      Handler執(zhí)行 task，讀取客戶端的請求數(shù)據(jù)（此處是非阻塞讀，即如果當前操作會導致當前線程阻塞，當前操作會立即返回，并重復執(zhí)行第 2、 3步，直到客戶端的請求讀取完畢。）

4)      解析客戶端請求數(shù)據(jù)

5)      讀取文件

6)      Handler重新綁定 write事件

7)      當 connection可以開始 write的時候， Reactor響應 write事件，分發(fā)給綁定了 write事件的 handler

8)    Handler 執(zhí)行 task ，向客戶端發(fā)送文件（此處是非阻塞寫，即如果當前操作會導致當前線程阻塞，當前操作會立即返回，并重復執(zhí)行第 7 、 8 步，直到文件全部發(fā)送完畢。）

注意：上述兩個過程都是在服務器的一個線程里完成的，該線程響應所有客戶端的請求。譬如服務端在處理客戶端 A 的請求時，如果在第 2 步 read 事件還沒有就緒（或者在第 3 步讀取數(shù)據(jù)的時候發(fā)生阻塞了），則該線程會立即重新回到客戶端連接服務器過程的第 2 步（即事件循環(huán)處理），等待新的事件通知。如果此時客戶端 B 請求連接，則該線程會響應 B 的連接請求，這樣就實現(xiàn)了一個線程同時為多個連接服務的效果。

3、 代碼示例

3.1 NIO的幾個關鍵概念

• Selector：

Reactor里的一個核心組成部分，通過調(diào)用 selector.select()方法，可以知道感興趣的 IO事件里哪些已經(jīng) ready，該方法是阻塞的，直到有 IO事件 ready；通過調(diào)用 selector.selectedKeys()方法，可以獲取到 selectionKey對象，這些對象關聯(lián)有已經(jīng) ready的 IO事件。

• SelectionKey：

當 selector注冊一個 channel時，會產(chǎn)生一個該對象，譬如SelectionKey selectionKey = channel .register(selector, SelectionKey. OP_ACCEPT );它維護著 channel 、 selector 、 IO 事件、 Handler 之間的關系。通過調(diào)用 attach 方法，可以綁定一個 handler ，譬如： selectionKey.attach(new Acceptor());

• ServerSocketChannel：

        類似于 ServerSocket，唯一的區(qū)別在于： ServerSocketChannel可以使用 selector，而且可以設置為非阻塞模式。

• SocketChannel：

        類似于 Socket，唯一的區(qū)別在于： SocketChannel可以使用 selector，而且可以設置為非阻塞模式。

• ByteBuffer ：數(shù)據(jù)緩沖器，是 NIO 里將數(shù)據(jù)移入移出 channel 的唯一方式

3.2 code

注：所有代碼只用來作為原理的進一步闡述，不能用于生產(chǎn)環(huán)境

Java代碼  

1. 服務端代碼如下（單線程版本）  
2.   
3. import java.io.IOException;  
4. import java.net.InetAddress;  
5. import java.net.InetSocketAddress;  
6. import java.nio.channels.SelectionKey;  
7. import java.nio.channels.Selector;  
8. import java.nio.channels.ServerSocketChannel;  
9. import java.nio.channels.SocketChannel;  
10. import java.nio.channels.spi.SelectorProvider;  
11. import java.util.Iterator;  
12.   
13.   
14. /** 
15.  * @author jason 
16.  *  
17.  */  
18. public class NioServer implements Runnable {  
19.   
20.     private InetAddress hostAddress;  
21.     private int port;  
22.   
23.     private ServerSocketChannel serverChannel;  
24.   
25.     private Selector selector;  
26.   
27.     public NioServer(InetAddress hostAddress, int port) throws IOException {  
28.         this.hostAddress = hostAddress;  
29.         this.port = port;  
30.         // 初始化selector，綁定服務端監(jiān)聽套接字、感興趣事件及對應的handler  
31.         this.selector = initSelector();  
32.     }  
33.   
34.     public static void main(String[] args) {  
35.         try {  
36.             // 啟動服務器  
37.             new Thread(new NioServer(null, 9090)).start();  
38.         } catch (IOException e) {  
39.             e.printStackTrace();  
40.         }  
41.     }  
42.   
43.     @Override  
44.     public void run() {  
45.         while (true) {  
46.             try {  
47.                 // 選擇事件已經(jīng)ready的selectionKey，該方法是阻塞的，只有當至少存在selectionKey，或者wakeup方法被調(diào)用，或者當前線程被中斷，才會返回  
48.                 selector.select();  
49.                 // 循環(huán)處理每一個事件  
50.                 Iterator<SelectionKey> items = selector.selectedKeys()  
51.                         .iterator();  
52.                 while (items.hasNext()) {  
53.                     SelectionKey key = (SelectionKey) items.next();  
54.                     items.remove();  
55.   
56.                     if (!key.isValid()) {  
57.                         continue;  
58.                     }  
59.                     // 事件處理分發(fā)  
60.                     dispatch(key);  
61.                 }  
62.   
63.             } catch (Exception e) {  
64.                 e.printStackTrace();  
65.             }  
66.         }  
67.   
68.     }  
69.   
70.     /** 
71.      * 事件處理分發(fā) 
72.      *  
73.      * @param sk 
74.      *            已經(jīng)ready的selectionKey 
75.      */  
76.     private void dispatch(SelectionKey sk) {  
77.         // 獲取綁定的handler  
78.         Runnable r = (Runnable) sk.attachment();  
79.         if (r != null) {  
80.             r.run();  
81.         }  
82.     }  
83.   
84.     /** 
85.      * 初始化selector，綁定服務端監(jiān)聽套接字、感興趣事件及對應的handler 
86.      *  
87.      * @return 
88.      * @throws IOException 
89.      */  
90.     private Selector initSelector() throws IOException {  
91.         // 創(chuàng)建一個selector  
92.         Selector socketSelector = SelectorProvider.provider().openSelector();  
93.         // 創(chuàng)建并打開ServerSocketChannel  
94.         serverChannel = ServerSocketChannel.open();  
95.         // 設置為非阻塞  
96.         serverChannel.configureBlocking(false);  
97.         // 綁定端口  
98.         serverChannel.socket().bind(new InetSocketAddress(hostAddress, port));  
99.         // 用selector注冊套接字，并返回對應的SelectionKey，同時設置Key的interest set為監(jiān)聽客戶端連接事件  
100.         SelectionKey selectionKey = serverChannel.register(socketSelector,  
101.                 SelectionKey.OP_ACCEPT);  
102.         // 綁定handler  
103.         selectionKey.attach(new Acceptor());  
104.   
105.         return socketSelector;  
106.     }  
107.   
108.     /** 
109.      * 處理OP_ACCEPT事件的handler 
110.      *  
111.      */  
112.     class Acceptor implements Runnable {  
113.         @Override  
114.         public void run() {  
115.             try {  
116.                 accept();  
117.             } catch (IOException e) {  
118.                 e.printStackTrace();  
119.             }  
120.   
121.         }  
122.   
123.         private void accept() throws IOException {  
124.             System.out.println("connect");  
125.             // 建立連接  
126.             SocketChannel socketChannel = serverChannel.accept();  
127.             System.out.println("connected");  
128.             // 設置為非阻塞  
129.             socketChannel.configureBlocking(false);  
130.             // 創(chuàng)建Handler,專門處理該連接后續(xù)發(fā)生的OP_READ和OP_WRITE事件  
131.             new Handler(selector, socketChannel);  
132.         }  
133.   
134.     }  
135.   
136. }  

Java代碼  

1. handler代碼如下  
2.   
3. import java.io.IOException;  
4. import java.nio.ByteBuffer;  
5. import java.nio.channels.SelectionKey;  
6. import java.nio.channels.Selector;  
7. import java.nio.channels.SocketChannel;  
8.   
9. /** 
10.  * @author jason 
11.  *  
12.  */  
13. final class Handler implements Runnable {  
14.     final SocketChannel socketChannel;  
15.     final SelectionKey key;  
16.     static final int MAXIN = 8192, MAXOUT = 11240 * 1024;  
17.     ByteBuffer readBuffer = ByteBuffer.allocate(MAXIN);  
18.     ByteBuffer outBuffer = ByteBuffer.allocate(MAXOUT);  
19.     static final int READING = 0;  
20.     static final int SENDING = 1;  
21.     int state = READING;  
22.   
23.     Handler(Selector selector, SocketChannel socketChannel) throws IOException {  
24.         this.socketChannel = socketChannel;  
25.         // 用selector注冊套接字，并返回對應的SelectionKey，同時設置Key的interest set為監(jiān)聽該連接上得read事件  
26.         this.key = socketChannel.register(selector, SelectionKey.OP_READ);  
27.         // 綁定handler  
28.         this.key.attach(this);  
29.     }  
30.   
31.     /** 
32.      * 處理write 
33.      *  
34.      * @throws IOException 
35.      */  
36.     private void write() throws IOException {  
37.         socketChannel.write(outBuffer);  
38.         if (outBuffer.remaining() > 0) {  
39.             return;  
40.         }  
41.         state = READING;  
42.         key.interestOps(SelectionKey.OP_READ);  
43.     }  
44.   
45.     /** 
46.      * 處理read 
47.      *  
48.      * @throws IOException 
49.      */  
50.     private void read() throws IOException {  
51.         readBuffer.clear();  
52.         int numRead;  
53.         try {  
54.             // 讀取數(shù)據(jù)  
55.             numRead = socketChannel.read(readBuffer);  
56.         } catch (Exception e) {  
57.             key.cancel();  
58.             socketChannel.close();  
59.             return;  
60.         }  
61.   
62.         if (numRead == -1) {  
63.             socketChannel.close();  
64.             key.cancel();  
65.             return;  
66.         }  
67.         // 處理數(shù)據(jù)  
68.         process(numRead);  
69.   
70.     }  
71.   
72.     /** 
73.      * 處理數(shù)據(jù) 
74.      *  
75.      * @param numRead 
76.      */  
77.     private void process(int numRead) {  
78.         byte[] dataCopy = new byte[numRead];  
79.         System.arraycopy(readBuffer.array(), 0, dataCopy, 0, numRead);  
80.         System.out.println(new String(dataCopy));  
81.         outBuffer = ByteBuffer.wrap(dataCopy);  
82.         state = SENDING;  
83.         // 設置Key的interest set為監(jiān)聽該連接上的write事件  
84.         key.interestOps(SelectionKey.OP_WRITE);  
85.     }  
86.   
87.     @Override  
88.     public void run() {  
89.         try {  
90.             if (state == READING) {  
91.                 read();  
92.             } else if (state == SENDING) {  
93.                 write();  
94.             }  
95.   
96.         } catch (IOException e) {  
97.             e.printStackTrace();  
98.         }  
99.   
100.     }  
101. }  

Java代碼  

1. 客戶端代碼如下：  
2.   
3. package sampleNio;  
4.   
5. import java.io.IOException;  
6. import java.net.InetAddress;  
7. import java.net.InetSocketAddress;  
8. import java.nio.ByteBuffer;  
9. import java.nio.channels.SelectionKey;  
10. import java.nio.channels.Selector;  
11. import java.nio.channels.SocketChannel;  
12. import java.nio.channels.spi.SelectorProvider;  
13. import java.util.Iterator;  
14.   
15. /** 
16.  * @author jason 
17.  *  
18.  */  
19. public class NioClient implements Runnable {  
20.     private InetAddress hostAddress;  
21.     private int port;  
22.     private Selector selector;  
23.     private ByteBuffer readBuffer = ByteBuffer.allocate(8192);  
24.     private ByteBuffer outBuffer = ByteBuffer.wrap("nice to meet you"  
25.             .getBytes());  
26.   
27.     public NioClient(InetAddress hostAddress, int port) throws IOException {  
28.         this.hostAddress = hostAddress;  
29.         this.port = port;  
30.         initSelector();  
31.     }  
32.   
33.     public static void main(String[] args) {  
34.         try {  
35.             NioClient client = new NioClient(  
36.                     InetAddress.getByName("localhost"), 9090);  
37.             new Thread(client).start();  
38.   
39.         } catch (IOException e) {  
40.             e.printStackTrace();  
41.         }  
42.     }  
43.   
44.     @Override  
45.     public void run() {  
46.         while (true) {  
47.             try {  
48.                 selector.select();  
49.   
50.                 Iterator<?> selectedKeys = selector.selectedKeys().iterator();  
51.                 while (selectedKeys.hasNext()) {  
52.                     SelectionKey key = (SelectionKey) selectedKeys.next();  
53.                     selectedKeys.remove();  
54.   
55.                     if (!key.isValid()) {  
56.                         continue;  
57.                     }  
58.   
59.                     if (key.isConnectable()) {  
60.                         finishConnection(key);  
61.                     } else if (key.isReadable()) {  
62.                         read(key);  
63.                     } else if (key.isWritable()) {  
64.                         write(key);  
65.                     }  
66.   
67.                 }  
68.   
69.             } catch (Exception e) {  
70.                 e.printStackTrace();  
71.             }  
72.         }  
73.   
74.     }  
75.   
76.     private void initSelector() throws IOException {  
77.         // 創(chuàng)建一個selector  
78.         selector = SelectorProvider.provider().openSelector();  
79.         // 打開SocketChannel  
80.         SocketChannel socketChannel = SocketChannel.open();  
81.         // 設置為非阻塞  
82.         socketChannel.configureBlocking(false);  
83.         // 連接指定IP和端口的地址  
84.         socketChannel  
85.                 .connect(new InetSocketAddress(this.hostAddress, this.port));  
86.         // 用selector注冊套接字，并返回對應的SelectionKey，同時設置Key的interest set為監(jiān)聽服務端已建立連接的事件  
87.         socketChannel.register(selector, SelectionKey.OP_CONNECT);  
88.     }  
89.   
90.     private void finishConnection(SelectionKey key) throws IOException {  
91.         SocketChannel socketChannel = (SocketChannel) key.channel();  
92.         try {  
93.             // 判斷連接是否建立成功，不成功會拋異常  
94.             socketChannel.finishConnect();  
95.         } catch (IOException e) {  
96.             key.cancel();  
97.             return;  
98.         }  
99.         // 設置Key的interest set為OP_WRITE事件  
100.         key.interestOps(SelectionKey.OP_WRITE);  
101.     }  
102.   
103.     /** 
104.      * 處理read 
105.      *  
106.      * @param key 
107.      * @throws IOException 
108.      */  
109.     private void read(SelectionKey key) throws IOException {  
110.         SocketChannel socketChannel = (SocketChannel) key.channel();  
111.         readBuffer.clear();  
112.         int numRead;  
113.         try {  
114.             numRead = socketChannel.read(readBuffer);  
115.         } catch (Exception e) {  
116.             key.cancel();  
117.             socketChannel.close();  
118.             return;  
119.         }  
120.         if (numRead == 1) {  
121.             System.out.println("close connection");  
122.             socketChannel.close();  
123.             key.cancel();  
124.             return;  
125.         }  
126.         // 處理響應  
127.         handleResponse(socketChannel, readBuffer.array(), numRead);  
128.     }  
129.   
130.     /** 
131.      * 處理響應 
132.      *  
133.      * @param socketChannel 
134.      * @param data 
135.      * @param numRead 
136.      * @throws IOException 
137.      */  
138.     private void handleResponse(SocketChannel socketChannel, byte[] data,  
139.             int numRead) throws IOException {  
140.         byte[] rspData = new byte[numRead];  
141.         System.arraycopy(data, 0, rspData, 0, numRead);  
142.         System.out.println(new String(rspData));  
143.         socketChannel.close();  
144.         socketChannel.keyFor(selector).cancel();  
145.     }  
146.   
147.     /** 
148.      * 處理write 
149.      *  
150.      * @param key 
151.      * @throws IOException 
152.      */  
153.     private void write(SelectionKey key) throws IOException {  
154.         SocketChannel socketChannel = (SocketChannel) key.channel();  
155.         socketChannel.write(outBuffer);  
156.         if (outBuffer.remaining() > 0) {  
157.             return;  
158.         }  
159.         // 設置Key的interest set為OP_READ事件  
160.         key.interestOps(SelectionKey.OP_READ);  
161.     }  
162.   
163. }  

4、 Reactor 的其他實現(xiàn)方式

單線程版本的 Reactor 最大的優(yōu)勢是：不需要做并發(fā)控制，簡化了實現(xiàn)。缺點是不能充分利用多核 CPU的優(yōu)勢，因為只有一個線程，該線程需要執(zhí)行所有的操作： accept、 read、 decode、 compute、 encode、 send，而其中的 decode、 compute、 encode如果很耗時，則該線程就不能及時的響應其他客戶端的請求。

為了解決該問題，可以采用另外兩種版本：

4.1 Worker threads：

Reactor所在的線程只需要專心的響應客戶端的請求： accept、 read、 send。對數(shù)據(jù)的具體處理過程則交給另外的線程池。這樣可以提高服務端對客戶端的響應速度，但同時增加了復雜度，也沒有充分利用到多核的優(yōu)勢，因為 reactor只有一個，譬如同一時刻只能 read一個客戶端的請求數(shù)據(jù)。

4.2Multiple reactor threads ：

采用多個 reactor ，每個 reactor 都在自己單獨的線程里執(zhí)行。如果是多核，則可以同時響應多個客戶端的請求。（ Netty 采用的是類似這種方式，boss線程池就是多個mainReactor，worker線程池就是多個subReactor）

5、總結(jié)

本文分析了基于 NIO和 Reactor模式的網(wǎng)絡服務器設計方案，在后續(xù)的 blog中將結(jié)合 Netty進一步分析高性能網(wǎng)絡服務器的設計。

 本文為原創(chuàng)，轉(zhuǎn)載請注明出處