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，可以看到都能得到正確回應。