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

Linux Network Programming - 初體驗

十幾年沒有寫 C 了，這可以說是重新學習，底下的程式是改寫自 W.Richard Stevens 的名著 - UNIX Network Programming Volume 1，改寫的原因有二:
(1) 在我的開發環境 scientific linux 7.0 上沒辦法正常 compile，可能是因為 Stevens 的程式是在 UNIX 上寫的，與 linux 上略有不同;
(2) Stevens 用 #define 將許多常用的函式重新定義，對於初學者來說，反而會被混淆，我將這些函式還原為原始函式。

這個程式有 client 和 server，server 先啟動，等待 client 連線，待 client 連線後回覆現在的系統時間，client 將收到的值印出。第一個程式是 server，第二個程式是 client。這裡會詳細說明每個函式，因為… 我是初學者...  XD

 1 #include <sys/socket.h>
 2 #include <sys/types.h>
 3 #include <stdio.h>
 4 #include <unistd.h>
 5 #include <netinet/in.h>
 6 #include <string.h>
 7 #include <arpa/inet.h>
 8 #include <time.h>
 9 
10 const int MAXLINE = 100;
11 const int LISTENQ = 1024;
12 
13 int main(int argc, char **argv) {
14     int listenfd, connfd;
15     struct sockaddr_in servaddr;
16     char buff[MAXLINE];
17 
18     time_t ticks;
19     listenfd = socket(AF_INET, SOCK_STREAM, 0);
20 
21     bzero(&servaddr, sizeof(servaddr));
22     servaddr.sin_family = AF_INET;
23     servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
24     servaddr.sin_port = htons(1513);
25 
26     bind(listenfd, (struct sockaddr *) &servaddr, sizeof(servaddr));
27     listen(listenfd, LISTENQ);
28 
29     for( ; ; ) {
30         connfd = accept(listenfd, (struct sockaddr *) NULL, NULL);
31 
32         ticks = time(NULL);
33         snprintf(buff, sizeof(buff), "%.24s\r\n", ctime(&ticks));
34         write(connfd, buff, strlen(buff));
35 
36         close(connfd);
37     }
38 
39 }

• 第 19 行: 建立一個 socket 連線，不管 client 或 server 程式，一開始都會呼叫這個函式，函式所需要的三個參數說明如下:
  • 第一個參數: AF_INET 表示要建立網際網路的 IPv4 socket 連線，除此之外還可以填入 AF_INET6 (IPv6) 或 AF_UNSPEC (同時可以用 IPv4 及 IPv6)，socket 函式及這幾個常數是定義在 socket.h 標頭檔裡，這也是為什麼第一行要引入這個標頭檔的原因，大部份的 socket 相關函式、常數都定義在這個標頭檔。
  • 第二個參數: 可以傳入的值有兩個 - SOCK_STREAM 及 SOCK_DGRAM，要建立 TCP 連線時傳入 SOCK_STREAM，要建立 UDP 連線時傳入 SOCK_DGRAM。
  • 第三個參數: 通常就是傳入 0，socket 將會選擇最適合的通訊協定，也就是第二個參數的說明，至於不傳入 0 時，用在什麼地方? 這個以後再說 (現在我也不知道)。
  • 傳回值: 錯誤時傳回 -1，正確傳回 0，C 語言的習慣就是正確傳回 0，錯誤則傳回錯誤代碼。
• 第 21 行: bzero 是將指定的位指空間全部清為 0，這不是 ANSI C 的標準函式，ANSI C 的標準函式是 memset，不過 bzero 大多數的平台也都有支援，定義在 string.h 裡。傳入的兩個參數如下:
  • 第一個參數: 要清為 0 的位址。
  • 第二個參數: 位址空間的長度。
• 第 22~24 行: 設定 server address 的值，這是在第 15 行宣告的變數，struct sockaddr_in 這個結構是為了在 bind 時傳入一些值，這三個值說明如下:
  • sin_family: 如 socket 的第二個參數說明的一樣，這是指要使用 TCP 通訊協定。
  • sin_addr.s_addr: 這裡指定為 INADDR_ANY 是使得任何位址連進來的 socket 連線都可以被接受，在設定前呼叫的 htonl 函式是為了讓值與平台無關，不同的 OS 有些字元排列順序高位元和低位元不一定相同，透過這個函式可以去除這種平台相依性，都轉為網路所規範的順序。
  • sin_port: 指定要 bind 到那個 port，htons 這個函式和 htonl 是一樣的，都是用來去除平台相依性，差別在於 htons 是用在 16 bits 的變數， htonl 是用在 32 bits 的變數。
• 第 26 行: bind 是將 socket 和 sockaddr 繫結起來的函式，表示第 19 行建立的 socket 連線，將會是允許任何位址連線，而將傾聽的 port 是 1513。
• 第 27 行: 開始傾聽，最大連線數是 LINTENQ 所定義的值，即 1024 條連線。
• 第 29 行: 這裡建立一個無窮迴圈，所以這個 server 會一直執行，直到使用者按 ctrl-C 為止。
• 第 30 行: server  會停在 accept 這裡等待 client 連線，參數說明如下:
  • 第一個參數: 透過傳入傾聽描述子 listenfd，將第 26、27 行傾聽的相關數據傳入。
  • 第二、三個參數: 待我弄懂了再補充  … XD
  • 傳回值: 建立連線後，傳回連線描述子 (connected descriptor)，這個描述子會用來與新的 client 連線進行通訊。
• 第 32 行: 取得目前的系統時間，這個 time 函式定義在 time.h 標頭檔裡。
• 第 33 行: snprintf 函式定義在 stdio.h 裡，和 sprintf 的最大差別在於 snprintf 的第二個參數指出了第一個參數的空間大小，可防止實際產生出來的字串長度超過第一個參數所宣告的空間。
• 第 34 行: 將結果寫出給 client，這裡可以看到第一個參數即是 accept 的傳回值，這樣就可以通知系統到底是要將資料傳給那個 client。
• 第 36 行: 關閉與 client 間的 socket 連線。

 1 #include <stdio.h>
 2 #include <sys/socket.h>
 3 #include <sys/types.h>
 4 #include <netinet/in.h>
 5 #include <strings.h>
 6 #include <arpa/inet.h>
 7 
 8 const int MAXLINE = 100;
 9 
10 void err_sys(const char* x, ...)
11 {
12     perror(x);
13     //exit(1);
14 }
15 
16 void err_quit(const char* x, ...)
17 {
18     perror(x);
19     //exit(1);
20 }
21 
22 int main(int argc, char **argv)
23 {
24     int sockfd, n;
25     char recvline[MAXLINE + 1];
26     struct sockaddr_in servaddr;
27 
28     if (argc != 2) {
29         err_quit("usage: a.out <IPaddress>");
30         return 1;
31     }
32 
33     if ((sockfd = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
34         err_sys("socket error");
35         return 1;
36     }
37 
38     bzero(&servaddr, sizeof(servaddr));
39     servaddr.sin_family = AF_INET;
40     servaddr.sin_port = htons(1513);
41     if (inet_pton(AF_INET, argv[1], &servaddr.sin_addr) <= 0)
42         err_quit("inet_pton error for %s", argv[1]);
43 
44     if (connect(sockfd, (struct sockaddr *) &servaddr, sizeof(servaddr)) < 0)
45         err_sys("connect error");
46 
47     while ((n = read(sockfd, recvline, MAXLINE)) > 0) {
48         recvline[n] = 0;
49         if (fputs(recvline, stdout) == EOF)
50             err_sys("fputs error");
51     }
52 
53     if (n < 0) {
54         err_sys("read error");
55         return 1;
56     }
57 
58     return 0;
59 }

• 第 41 行: client 程式執行時，要帶入 server 的 IP，inet_pton 函式即是將 IP 的文字格式轉成接下來要呼叫的 connect 所需要的 binary 格式。
• 第 44 行: 呼叫 connect 連線到 server。
• 第 47 行: 使用 read 讀取 server 傳回的值，三個參數說明如下:
  • 第一個參數: socket 描述子，第 33 行建立 socket 連線時記錄下來的數值。
  • 第二個參數: 讀取的值要放入的空間。
  • 第三個參數: 放入的空間的最大長度值。
• 第 49 行: 將結果寫出到標準輸出裝置。

測試時，先執行 server 再執行 client，client 端顯示如下結果:

【日劇: Second Love】
在台灣播出時改名為「愛上女老師」，看這部日劇前，沒特別注意過深田恭子原來那麼偉大，而且都年過三十了，還那麼可愛，算是不簡單，許多人上了年紀還裝可愛，可是會被吐槽的，但是她不會。這部戲為什麼只有七集? 是因為收視率太低嗎? 這就不曉得了，但是除了結局太老套外，倒是還不錯看。