NIO.2: TCP 網路程式設計 (non-blocking)

上一篇說明了 blocking 程式，這一篇要說明如何寫 non-blocking 程式，首先看一下兩者的差別:

• blocking 模式一個 thread 只能有一個 channel，所以，server 端要服務多個 client 的話，就要為每個 client 建立一個 thread。
• non-blocking 模式引進一個新的類別 - Selector，這個類別的用處在於，讓 non-blocking 成為事件驅動的運作模式。在 blocking 模式下，都是由程式本身決定何時連線與讀寫，這造成的限制就是當要同時有多個連線同時運作時，就要有多個 thread; Selector 會從 ServerSocketChannel 中選擇一個目前需要服務的 channel 進行服務，傳給程式 channel 及這個 channel 需要什麼類型的服務。

了解了 non-blocking 的運作方式後，要先說明另一個類別 - SelectionKey，這個類別中定義了四個運作模式 (也可視為四種事件)，也就是 Selector 要告訴程式，這個 channel 需要什麼服務，如下:

• SelectionKey.OP_ACCEPT: 這事件通常是用在 server 端，當 client 連線到 server 時，產生這個運作模式。
• SelectionKey.OP_CONNECT: 這通常用在 client 端，當 server 端接受了 client 端的連線，client 端就會收到這個 event。
• SelectionKey.OP_READ: 當有資料進來，可以讀的時候，會收到這個事件。
• SelectionKey.OP_WRITE: 當收到這個事件，即可以寫出資料到該事件所關聯的 channel。

要看程式前，還是要先說明，non-blocking 是建立在 blocking 的基礎上，增加的這些類別，是為將運作方式轉換成事件驅動，由 ServerSocketChannel 傳訊息給 Selector，再由 Selector 傳給程式，了解這些觀念之後再來看下面的程式，應該就很容易了解了。
(感覺 non-blocking 有點像 IoC，把程式的主動權交出去，反而讓程式得到更多彈性。)

 1 package idv.steven.async;
  2 
  3 import java.io.IOException;
  4 import java.net.InetSocketAddress;
  5 import java.net.StandardSocketOptions;
  6 import java.nio.ByteBuffer;
  7 import java.nio.channels.SelectionKey;
  8 import java.nio.channels.Selector;
  9 import java.nio.channels.ServerSocketChannel;
 10 import java.nio.channels.SocketChannel;
 11 import java.util.Arrays;
 12 import java.util.Iterator;
 13 import java.util.Set;
 14 import java.util.concurrent.ExecutorService;
 15 import java.util.concurrent.Executors;
 16 
 17 public class EchoServer2 implements Runnable {
 18     public static int DEFAULT_PORT = 5555;
 19 
 20     @Override
 21     public void run() {
 22         ServerSocketChannel serverChannel;
 23         Selector selector;
 24         
 25         try {
 26             serverChannel = ServerSocketChannel.open();
 27             selector = Selector.open();
 28             
 29             if (serverChannel.isOpen() && selector.isOpen()) {
 30                 // 設定為 false 即表示要用 non-blocking 模式
 31                 serverChannel.configureBlocking(false);
 32                 
 33                 serverChannel.setOption(StandardSocketOptions.SO_RCVBUF, 256 * 1024);
 34                 serverChannel.setOption(StandardSocketOptions.SO_REUSEADDR, true);
 35                 
 36                 serverChannel.bind(new InetSocketAddress(DEFAULT_PORT));
 37                 // 註冊 OP_ACCEPT 給 ServerSocketChannel，這樣 ServerSocketChannel 接到 client 端連線時，就會有一個值為 OP_ACCEPT 的 SelectionKey。
 39                 serverChannel.register(selector, SelectionKey.OP_ACCEPT);
 40                 
 41                 while (true) {
 42                     // 程式會在這裡停下，等待關心的"selection operation"出現。這個 method 有另一個版本，可以傳入 timeout 值，以免程式一直停在這裡。
 44                     selector.select();
 45                     
 46                     Set<SelectionKey> readyKeys = selector.selectedKeys();
 47                     Iterator<SelectionKey> iterator = readyKeys.iterator();
 48                     while (iterator.hasNext()) {
 49                         SelectionKey key = iterator.next();
 50                         // 移除將處理的 selection operation 很重要!
 51                         // 避免之後又重複接收到。
 52                         iterator.remove();
 53                         
 54                         try {
 55                             if (key.isAcceptable()) {
 56                                 // ServerSocketChannel 的用處就只是接受 client 連線，當收到 OP_ACCEPT 時，表示有 client 要連線，所以用 ServerSocketChannel 接收。
 58                                 ServerSocketChannel server = (ServerSocketChannel) key.channel();
 59                                 // 等待 client 連線
 60                                 SocketChannel client = server.accept();
 61                                 // 連線後設定這個連線為 non-blocking
 62                                 client.configureBlocking(false);
 63                                 // 這是一個 echo server，連線後緊接著要收 client 送來的訊息，所以註冊這個 socket channel 要關注 OP_READ。
 65                                 SelectionKey clientKey = client.register(selector, SelectionKey.OP_READ);
 66                                 ByteBuffer buffer = ByteBuffer.allocate(1024);
 67                                 // non-blocking 模式有點像事件模式，每次都是待 selection operation 出現，程式才會接手做相關的事，在這不同的 selection operation 之間，要如何保留讀入的資料? 使用 attach() 及 attachment()，讓這些資料保存在 SelectionKey 中。 
 71                                 clientKey.attach(buffer);
 72                             }
 73                             else if (key.isReadable()) {
 74                                 SocketChannel client = (SocketChannel) key.channel();
 75 //在 OP_ACCEPT 階段，程式有預先產生一個 ByteBuffer，這時將它取出，用來儲存讀入的資料。
 77                                 ByteBuffer output = (ByteBuffer) key.attachment();
 78                                 client.read(output);
 79                                 System.out.println("recv: " + new String(Arrays.copyOfRange(output.array(), 0, output.limit())));
 80                                 // echo server 讀到資料後，當然接著要寫回給 client，所以註冊 OP_WRITE。
 81                                 SelectionKey clientKey = client.register(selector, SelectionKey.OP_WRITE);
 82                             }
 83                             else if (key.isWritable()) {
 84                                 SocketChannel client = (SocketChannel) key.channel();
 85                                 // 在 OP_READ 階段讀取到的資料，現在將它取出。
 86                                 ByteBuffer output = (ByteBuffer) key.attachment();
 87                                 if (output != null) {
 88                                     output.flip();
 89                                     if (output.hasRemaining()) {
 90                                         // 把讀取到的資料寫回給 client。
 91                                         client.write(output);
 92                                         output.compact();
 93                                     }
 94                                     else {
 95                                         System.out.println("output has not remaining");
 96                                     }
 97                                 }
 98                             }
 99                         } catch (IOException e) {
100                             key.cancel();
101                             try {
102                                 key.channel().close();
103                             } catch (IOException ex) { }
104                         }
105                     }
106                 }
107             }
108         } catch (IOException e) {
109             e.printStackTrace();
110         }
111     }
112     
113     public static void main(String[] args) {
114         ExecutorService svc = Executors.newSingleThreadExecutor();
115         EchoServer2 echo = new EchoServer2();
116         svc.execute(echo);
117         svc.shutdown();
118     }
119 }

使用上一篇的 client 來測試，得到如下結果:

我開兩個視窗，編寫了批次程式，讓兩邊都有 client 程式一直送資料給 server，可以看到都能得到正確回應。

NIO.2: TCP 網路程式設計 (blocking)

這一篇要說明的是，如何使用 JDK 7 推出的 NIO 2 進行 TCP 程式開發，底下的類別圖只是其中較重要的部份，要詳細了解各類別關係，及各類別的 method，請查閱 API Documents。

NIO 2 提供 blocking (阻斷)、non-blocking (非阻斷) 及 asynchronous (非同步) 三種模式，其中 asynchronous 模式是到 Java 7 才新提供的，我還沒有研究，先不說明。 用 blocking 模式和 non-blocking 模式來寫 server 的話，在一開始 blocking 會停在 accept 而 non-blocking 會停在 select 等待 client 的連線，看起來兩者使用的類別雖有不同，行為模式卻相同，但自此之後兩者就開始不一樣了。至於有什麼不一樣，待討論到 non-blocking 模式時會作說明，現在只要了解，blocking 模式的程式，如果 server 要同時服務多個 client，一個 channel 會有一個 thread，而 non-blocking 模式則是在一個 thread 中服務所有的 client。底下先來看一個 blocking 程式，很簡單的一個 echo 程式，client 傳送隨機的數字給 server，server 於字串前加上"回傳:"後傳回給 client，執行結果如下: (我總共執行了三次)

接下來看程式及說明:

• client

 1 package idv.steven.sync;
 2 
 3 import java.io.IOException;
 4 import java.net.InetSocketAddress;
 5 import java.net.StandardSocketOptions;
 6 import java.nio.ByteBuffer;
 7 import java.nio.CharBuffer;
 8 import java.nio.channels.SocketChannel;
 9 import java.nio.charset.Charset;
10 import java.nio.charset.CharsetDecoder;
11 import java.util.Random;
12 
13 public class Client {
14 
15     public static void main(String[] args) {
16         final int DEFAULT_PORT = 5555;
17         final String IP = "127.0.0.1";
18 
19         ByteBuffer rcvBuffer = ByteBuffer.allocateDirect(1024);
20         Charset charset = Charset.forName("UTF-8"); // 網路傳輸時以 UTF-8 編碼
21         CharsetDecoder decoder = charset.newDecoder();
22         
23         // 建立 socket channel，在 NIO 2 裡，讀寫由 stream 改為 channel。
24         try (SocketChannel socketChannel = SocketChannel.open()) {
25             // 確認開啟 channel 成功
26             if (socketChannel.isOpen()) {
27                 // NIO 2 為 TCP 網路程式提供 blocking 和 non-blocking 兩種模式
28                 socketChannel.configureBlocking(true);
29                 // 設定 socket 的一些參數
30                 socketChannel.setOption(StandardSocketOptions.SO_RCVBUF, 128 * 1024);
31                 socketChannel.setOption(StandardSocketOptions.SO_SNDBUF, 128 * 1024);
32                     // 要求 JVM 保持 socket 連線，不過，要如何保持和 OS 有關，還是得 OS 決定。
33                 socketChannel.setOption(StandardSocketOptions.SO_KEEPALIVE, true);
34                     // 當 channel close 時，還有未送出的資料，要等待幾秒? 時間到仍強制關閉!
35                 socketChannel.setOption(StandardSocketOptions.SO_LINGER, 5);
36                 
37                 // 連線到 server socket
38                 socketChannel.connect(new InetSocketAddress(IP, DEFAULT_PORT));
39                 
40                 // 確認連線是否成功
41                 if (socketChannel.isConnected()) {
42                     Random random = new Random();
43                     ByteBuffer sendBuffer = ByteBuffer.wrap(String.valueOf(random.nextInt(100)).getBytes());
44                      socketChannel.write(sendBuffer);
45                     // socket 讀資料，不一定一次就可以完全讀完對方送來的資料，要讀到沒資料為止。
46                     while (socketChannel.read(rcvBuffer) != -1) {
47                         System.out.println("continue ...");
48                      }
49                      rcvBuffer.flip();
50                     CharBuffer charBuffer = decoder.decode(rcvBuffer);
51                      System.out.println(charBuffer);
52                 } else {
53                     System.out.println("連線失敗!");
54                 }
55             } else {
56                 System.out.println("socket channel 開啟失敗!");
57             }
58         } catch (IOException ex) {
59             System.err.println(ex);
60         }
61     }
62 }

• server

 1 package idv.steven.sync;
 2 
 3 import java.io.IOException;
 4 import java.net.InetSocketAddress;
 5 import java.net.StandardSocketOptions;
 6 import java.nio.ByteBuffer;
 7 import java.nio.CharBuffer;
 8 import java.nio.channels.ServerSocketChannel;
 9 import java.nio.channels.SocketChannel;
10 import java.nio.charset.Charset;
11 import java.nio.charset.CharsetDecoder;
12 import java.nio.charset.CharsetEncoder;
13 
14 public class EchoServer {
15  
16     public static void main(String[] args) {
17         final int DEFAULT_PORT = 5555;
18         final String IP = "127.0.0.1";
19         ByteBuffer buffer = ByteBuffer.allocateDirect(1024);
20         Charset charset = Charset.forName("UTF-8"); // 網路傳輸時以 UTF-8 編碼
21         CharsetEncoder encoder = charset.newEncoder();
22         CharsetDecoder decoder = charset.newDecoder();
23         
24         // 建立一個 server socket channel
25         try (ServerSocketChannel serverSocketChannel = ServerSocketChannel.open()) {
26             // 確認 server socket channel 開啟成功
27             if (serverSocketChannel.isOpen()) {
28                 // 設定 TCP 傳輸方式為 blocking 模式
29                 serverSocketChannel.configureBlocking(true);
30                 // 設定 channel 的一些參數
31                 serverSocketChannel.setOption(StandardSocketOptions.SO_RCVBUF, 4 * 1024);
32                 serverSocketChannel.setOption(StandardSocketOptions.SO_REUSEADDR, true);
33                 // 繫結 server socket 到指定的 IP 和 port
34                 serverSocketChannel.bind(new InetSocketAddress(IP, DEFAULT_PORT));
35 
36                 while(true){
37                     System.out.println("等待連線 ...");
38                     try (SocketChannel socketChannel = serverSocketChannel.accept()) {
39                         System.out.println("連線來至: " + socketChannel.getRemoteAddress());
40                         while (socketChannel.read(buffer) <= 0) {
41                             System.out.println("continue ...");
42                          }
43                         
44                          buffer.flip();
45                         CharBuffer charBuffer = CharBuffer.wrap("回傳: " + decoder.decode(buffer));
46                         ByteBuffer response = encoder.encode(charBuffer);
47                          socketChannel.write(response);
48                          buffer.clear();
49                     } 
50                    catch (IOException ex) {
51                         ex.printStackTrace();
52                     }
53                 }
54             } else {
55                 System.out.println("server socket channel 開啟失數!");
56             }
57         } 
58         catch (IOException ex) {
59             System.err.println(ex);
60         }
61     }
62 }

上面的程式，client 和 server 間傳遞的訊息(電文)就只有個數字，一般來說，TCP 程式的電文不會這麼簡單，大部份會有 header 和 body，在電文最前面也會有個特殊字元當作起始，電文最後還會有特殊字元當作結尾。在台灣證券交易所的網站，就有台股相關的電文規格書，是個很好的參考，進到網站後，選擇上方選單「市場公告 > 公文查詢」，找到「資訊傳輸作業手冊」就有完整的規格。