http://blog.csdn.net/shagoo/article/details/6396089

<?php
/**
 * SocketServer Class
 * By James.Huang <shagoo#gmail.com>
**/
set_time_limit(0);
class SocketServer 
{
    private static $socket;
    function SocketServer($port) 
    {
        global $errno, $errstr;
        if ($port < 1024) {
            die("Port must be a number which bigger than 1024/n");
        }
        
        $socket = stream_socket_server("tcp://0.0.0.0:{$port}", $errno, $errstr);
        if (!$socket) die("$errstr ($errno)");
        
//        stream_set_timeout($socket, -1); // 保證服務端 socket 不會超時，似乎沒用：）
        
        while ($conn = stream_socket_accept($socket, -1)) { // 這樣設置不超時才油用
            static $id = 0;
            static $ct = 0;
            $ct_last = $ct;
            $ct_data = '';
            $buffer = '';
            $id++; // increase on each accept
            echo "Client $id come./n";
            while (!preg_match('//r?/n/', $buffer)) { // 沒有讀到結束符，繼續讀
//                if (feof($conn)) break; // 防止 popen 和 fread 的 bug 導致的死循環
                $buffer = fread($conn, 1024);
                echo 'R'; // 打印讀的次數
                $ct += strlen($buffer);
                $ct_data .= preg_replace('//r?/n/', '', $buffer);
            }
            $ct_size = ($ct - $ct_last) * 8;
            echo "[$id] " . __METHOD__ . " > " . $ct_data . "/n";
            fwrite($conn, "Received $ct_size byte data./r/n");
            fclose($conn);
        }
        
        fclose($socket);
    }
}
new SocketServer(2000);

<?php
/**
 * Socket Test Client
 * By James.Huang <shagoo#gmail.com>
**/
function debug ($msg)
{
//    echo $msg;
    error_log($msg, 3, '/tmp/socket.log');
}
if ($argv[1]) {
    
    $socket_client = stream_socket_client('tcp://0.0.0.0:2000', $errno, $errstr, 30);
    
//    stream_set_blocking($socket_client, 0);
//    stream_set_timeout($socket_client, 0, 100000);
    
    if (!$socket_client) {
        die("$errstr ($errno)");
    } else {
        $msg = trim($argv[1]);
        for ($i = 0; $i < 10; $i++) {
            $res = fwrite($socket_client, "$msg($i)");
            usleep(100000);
            echo 'W'; // 打印寫的次數
//            debug(fread($socket_client, 1024)); // 將產生死鎖，因為 fread 在阻塞模式下未讀到數據時將等待
        }
        fwrite($socket_client, "/r/n"); // 傳輸結束符
        debug(fread($socket_client, 1024));
        fclose($socket_client);
    }
}
else {
    
//    $phArr = array();
//    for ($i = 0; $i < 10; $i++) {
//        $phArr[$i] = popen("php ".__FILE__." '{$i}:test'", 'r');
//    }
//    foreach ($phArr as $ph) {
//        pclose($ph);
//    }
    
    for ($i = 0; $i < 10; $i++) {
        system("php ".__FILE__." '{$i}:test'");
    }
}

<?php
/**
 * SelectSocketServer Class
 * By James.Huang <shagoo#gmail.com>
**/
set_time_limit(0);
class SelectSocketServer 
{
    private static $socket;
    private static $timeout = 60;
    private static $maxconns = 1024;
    private static $connections = array();
    function SelectSocketServer($port) 
    {
        global $errno, $errstr;
        if ($port < 1024) {
            die("Port must be a number which bigger than 1024/n");
        }
        
        $socket = socket_create_listen($port);
        if (!$socket) die("Listen $port failed");
        
        socket_set_nonblock($socket); // 非阻塞
        
        while (true) 
        {
            $readfds = array_merge(self::$connections, array($socket));
            $writefds = array();
            
            // 選擇一個連接，獲取讀、寫連接通道
            if (socket_select($readfds, $writefds, $e = null, $t = self::$timeout)) 
            {
                // 如果是當前服務端的監聽連接
                if (in_array($socket, $readfds)) {
                    // 接受客戶端連接
                    $newconn = socket_accept($socket);
                    $i = (int) $newconn;
                    $reject = '';
                    if (count(self::$connections) >= self::$maxconns) {
                        $reject = "Server full, Try again later./n";
                    }
                    // 將當前客戶端連接放入 socket_select 選擇
                    self::$connections[$i] = $newconn;
                    // 輸入的連接資源緩存容器
                    $writefds[$i] = $newconn;
                    // 連接不正常
                    if ($reject) {
                        socket_write($writefds[$i], $reject);
                        unset($writefds[$i]);
                        self::close($i);
                    } else {
                        echo "Client $i come./n";
                    }
                    // remove the listening socket from the clients-with-data array
                    $key = array_search($socket, $readfds);
                    unset($readfds[$key]);
                }
                
                // 輪循讀通道
                foreach ($readfds as $rfd) {
                    // 客戶端連接
                    $i = (int) $rfd;
                    // 從通道讀取
                    $line = @socket_read($rfd, 2048, PHP_NORMAL_READ);
                    if ($line === false) {
                        // 讀取不到內容，結束連接          
                        echo "Connection closed on socket $i./n";
                        self::close($i);
                        continue;
                    }
                    $tmp = substr($line, -1);
                    if ($tmp != "/r" && $tmp != "/n") {
                        // 等待更多數據
                        continue;
                    }
                    // 處理邏輯
                    $line = trim($line);
                    if ($line == "quit") {
                        echo "Client $i quit./n";
                        self::close($i);
                        break;
                    }
                    if ($line) {
                        echo "Client $i >>" . $line . "/n";
                    }
                }
                
                // 輪循寫通道
                foreach ($writefds as $wfd) {
                    $i = (int) $wfd;
                    $w = socket_write($wfd, "Welcome Client $i!/n");
                }
            }
        }
    }
    
    function close ($i) 
    {
        socket_shutdown(self::$connections[$i]);
        socket_close(self::$connections[$i]);
        unset(self::$connections[$i]);
    }
}
new SelectSocketServer(2000);

<?php
/**
 * SelectSocket Test Client
 * By James.Huang <shagoo#gmail.com>
**/
function debug ($msg)
{
//    echo $msg;
    error_log($msg, 3, '/tmp/socket.log');
}
if ($argv[1]) {
    
    $socket_client = stream_socket_client('tcp://0.0.0.0:2000', $errno, $errstr, 30);
    
//    stream_set_timeout($socket_client, 0, 100000);
    
    if (!$socket_client) {
        die("$errstr ($errno)");
    } else {
        $msg = trim($argv[1]);
        for ($i = 0; $i < 10; $i++) {
            $res = fwrite($socket_client, "$msg($i)/n");
            usleep(100000);
//            debug(fread($socket_client, 1024)); // 將產生死鎖，因為 fread 在阻塞模式下未讀到數據時將等待
        }
        fwrite($socket_client, "quit/n"); // add end token
        debug(fread($socket_client, 1024));
        fclose($socket_client);
    }
}
else {
    
    $phArr = array();
    for ($i = 0; $i < 10; $i++) {
        $phArr[$i] = popen("php ".__FILE__." '{$i}:test'", 'r');
    }
    foreach ($phArr as $ph) {
        pclose($ph);
    }
    
//    for ($i = 0; $i < 10; $i++) {
//        system("php ".__FILE__." '{$i}:test'");
//    }
}

<?php
/**
 * EpollSocketServer Class (use libevent)
 * By James.Huang <shagoo#gmail.com>
 * 
 * Defined constants:
 * 
 * EV_TIMEOUT (integer)
 * EV_READ (integer)
 * EV_WRITE (integer)
 * EV_SIGNAL (integer)
 * EV_PERSIST (integer)
 * EVLOOP_NONBLOCK (integer)
 * EVLOOP_ONCE (integer)
**/
set_time_limit(0);
class EpollSocketServer
{
    private static $socket;
    private static $connections;
    private static $buffers;
    
    function EpollSocketServer ($port)
    {
        global $errno, $errstr;
        
        if (!extension_loaded('libevent')) {
            die("Please install libevent extension firstly/n");
        }
        
        if ($port < 1024) {
            die("Port must be a number which bigger than 1024/n");
        }
        
        $socket_server = stream_socket_server("tcp://0.0.0.0:{$port}", $errno, $errstr);
        if (!$socket_server) die("$errstr ($errno)");
        
        stream_set_blocking($socket_server, 0); // 非阻塞
        
        $base = event_base_new();
        $event = event_new();
        event_set($event, $socket_server, EV_READ | EV_PERSIST, array(__CLASS__, 'ev_accept'), $base);
        event_base_set($event, $base);
        event_add($event);
        event_base_loop($base);
        
        self::$connections = array();
        self::$buffers = array();
    }
    
    function ev_accept($socket, $flag, $base) 
    {
        static $id = 0;
    
        $connection = stream_socket_accept($socket);
        stream_set_blocking($connection, 0);
    
        $id++; // increase on each accept
    
        $buffer = event_buffer_new($connection, array(__CLASS__, 'ev_read'), array(__CLASS__, 'ev_write'), array(__CLASS__, 'ev_error'), $id);
        event_buffer_base_set($buffer, $base);
        event_buffer_timeout_set($buffer, 30, 30);
        event_buffer_watermark_set($buffer, EV_READ, 0, 0xffffff);
        event_buffer_priority_set($buffer, 10);
        event_buffer_enable($buffer, EV_READ | EV_PERSIST);
    
        // we need to save both buffer and connection outside
        self::$connections[$id] = $connection;
        self::$buffers[$id] = $buffer;
    }
    
    function ev_error($buffer, $error, $id) 
    {
        event_buffer_disable(self::$buffers[$id], EV_READ | EV_WRITE);
        event_buffer_free(self::$buffers[$id]);
        fclose(self::$connections[$id]);
        unset(self::$buffers[$id], self::$connections[$id]);
    }
    
    function ev_read($buffer, $id) 
    {
        static $ct = 0;
        $ct_last = $ct;
        $ct_data = '';
        while ($read = event_buffer_read($buffer, 1024)) {
            $ct += strlen($read);
            $ct_data .= $read;
        }
        $ct_size = ($ct - $ct_last) * 8;
        echo "[$id] " . __METHOD__ . " > " . $ct_data . "/n";
        event_buffer_write($buffer, "Received $ct_size byte data./r/n");
    }
    
    function ev_write($buffer, $id) 
    {
        echo "[$id] " . __METHOD__ . "/n";
    }
}
new EpollSocketServer(2000);

<?php
/**
 * EpollSocket Test Client
 * By James.Huang <shagoo#gmail.com>
**/
function debug ($msg)
{
//    echo $msg;
    error_log($msg, 3, '/tmp/socket.log');
}
if ($argv[1]) {
    $socket_client = stream_socket_client('tcp://0.0.0.0:2000', $errno, $errstr, 30);
//    stream_set_blocking($socket_client, 0);
    if (!$socket_client) {
        die("$errstr ($errno)");
    } else {
        $msg = trim($argv[1]);
        for ($i = 0; $i < 10; $i++) {
            $res = fwrite($socket_client, "$msg($i)");
            usleep(100000);
            debug(fread($socket_client, 1024));
        }
        fclose($socket_client);
    }
}
else {
    
    $phArr = array();
    for ($i = 0; $i < 10; $i++) {
        $phArr[$i] = popen("php ".__FILE__." '{$i}:test'", 'r');
    }
    foreach ($phArr as $ph) {
        pclose($ph);
    }
    
//    for ($i = 0; $i < 10; $i++) {
//        system("php ".__FILE__." '{$i}:test'");
//    }
}

http://blog.csdn.net/shagoo/article/details/6531950

上一篇《Socket深度探究4PHP（一）》中，大家應該對 poll/select/epoll/kqueue 這幾個 IO 模型有了一定的了解，為了讓大家更深入的理解 Socket 的技術內幕，在這個篇幅，我會對這幾種模式做一個比較詳細的分析和對比；另外，大家可能也同說過 AIO 的概念，這里也會做一個簡單的介紹；最后我們會對兩種主流異步模式 Reactor 和 Proactor 模式進行對比和討論。

首先，然我們逐個介紹一下 2.6 內核（2.6.21.1）中的 poll/select/epoll/kqueue 這幾個 IO 模型。

> POLL

先說說 poll，poll 和 select 為大部分 Unix/Linux 程序員所熟悉，這倆個東西原理類似，性能上也不存在明顯差異，但 select 對所監控的文件描述符數量有限制，所以這里選用 poll 做說明。poll 是一個系統調用，其內核入口函數為 sys_poll，sys_poll 幾乎不做任何處理直接調用 do_sys_poll，do_sys_poll 的執行過程可以分為三個部分：

1、將用戶傳入的 pollfd 數組拷貝到內核空間，因為拷貝操作和數組長度相關，時間上這是一個 O(n) 操作，這一步的代碼在 do_sys_poll 中包括從函數開始到調用 do_poll 前的部分。

2、查詢每個文件描述符對應設備的狀態，如果該設備尚未就緒，則在該設備的等待隊列中加入一項并繼續查詢下一設備的狀態。查詢完所有設備后如果沒有一個設備就緒，這時則需要掛起當前進程等待，直到設備就緒或者超時，掛起操作是通過調用 schedule_timeout 執行的。設備就緒后進程被通知繼續運行，這時再次遍歷所有設備，以查找就緒設備。這一步因為兩次遍歷所有設備，時間復雜度也是 O(n)，這里面不包括等待時間。相關代碼在 do_poll 函數中。

3、將獲得的數據傳送到用戶空間并執行釋放內存和剝離等待隊列等善后工作，向用戶空間拷貝數據與剝離等待隊列等操作的的時間復雜度同樣是 O(n)，具體代碼包括 do_sys_poll 函數中調用 do_poll 后到結束的部分。

但是，即便通過 select() 或者 poll() 函數復用事件通知具有突出的優點，不過其他具有類似功能的函數實現也可以達到同樣的性能。然而，這些實現在跨平臺方面沒有實現標準化。你必須在使用這些特定函數實現同喪失可移植性之間進行權衡。我們現在就討論一下兩個替代方法：Solaris 系統下的 /dev/poll 和 FreeBSD 系統下的 kqueue：

1、Solaris 系統下的 /dev/poll：在Solaris 7系統上，Sun引入了/dev/poll設備。在使用 /dev/poll的時候，你首先要打開/dev/poll作為一個普通文件。然后構造pollfd結構，方式同普通的poll()函數調用一樣。這些 pollfd結構隨后寫入到打開的 /dev/poll 文件描述符。在打開句柄的生存周期內， /dev/poll會根據pollfd結構返回事件（注意，pollfd結構內的事件字段中的特定POLLREMOVE將從/dev/poll的列表中刪除對應的fd）。通過調用特定的ioctl (DP_POLL) 和dvpoll，程序就可以從/dev/poll獲得需要的信息。在使用dvpoll結構的情況下，發生的事件就可以被檢測到了。

2、FreeBSD 系統下的 kqueue：在FreeBSD 4.1中推出。FreeBSD的kqueue API設計為比其他對應函數提供更為廣泛的事件通知能力。kqueue API提供了一套通用過濾器，可以模仿poll()語法（EVFILT_READ和EVFILT_WRITE）。不過，它還實現了文件系統變化（EVFILT_VNODE）、進程狀態變更(EVFILT_PROC)和信號交付(EVFILT_SIGNAL)的有關通知。

> EPOLL

接下來分析 epoll，與 poll/select 不同，epoll 不再是一個單獨的系統調用，而是由 epoll_create/epoll_ctl/epoll_wait 三個系統調用組成，后面將會看到這樣做的好處。先來看 sys_epoll_create（epoll_create對應的內核函數），這個函數主要是做一些準備工作，比如創建數據結構，初始化數據并最終返回一個文件描述符（表示新創建的虛擬 epoll 文件），這個操作可以認為是一個固定時間的操作。epoll 是做為一個虛擬文件系統來實現的，這樣做至少有以下兩個好處：

1、可以在內核里維護一些信息，這些信息在多次 epoll_wait 間是保持的，比如所有受監控的文件描述符。

2、epoll 本身也可以被 poll/epoll。

具體 epoll 的虛擬文件系統的實現和性能分析無關，不再贅述。

在 sys_epoll_create 中還能看到一個細節，就是 epoll_create 的參數 size 在現階段是沒有意義的，只要大于零就行。

接著是 sys_epoll_ctl（epoll_ctl對應的內核函數），需要明確的是每次調用 sys_epoll_ctl 只處理一個文件描述符，這里主要描述當 op 為 EPOLL_CTL_ADD 時的執行過程，sys_epoll_ctl 做一些安全性檢查后進入 ep_insert，ep_insert 里將 ep_poll_callback 做為回掉函數加入設備的等待隊列（假定這時設備尚未就緒），由于每次 poll_ctl 只操作一個文件描述符，因此也可以認為這是一個 O(1) 操作。ep_poll_callback 函數很關鍵，它在所等待的設備就緒后被系統回掉，執行兩個操作：

1、將就緒設備加入就緒隊列，這一步避免了像 poll 那樣在設備就緒后再次輪詢所有設備找就緒者，降低了時間復雜度，由 O(n) 到 O(1)。
 
2、喚醒虛擬的 epoll 文件。

最后是 sys_epoll_wait，這里實際執行操作的是 ep_poll 函數。該函數等待將進程自身插入虛擬 epoll 文件的等待隊列，直到被喚醒（見上面 ep_poll_callback 函數描述），最后執行 ep_events_transfer 將結果拷貝到用戶空間。由于只拷貝就緒設備信息，所以這里的拷貝是一個 O(1) 操作。

還有一個讓人關心的問題就是 epoll 對 EPOLLET 的處理，即邊沿觸發的處理，粗略看代碼就是把一部分水平觸發模式下內核做的工作交給用戶來處理，直覺上不會對性能有太大影響，感興趣的朋友歡迎討論。

> POLL/EPOLL 對比：

表面上 poll 的過程可以看作是由一次 epoll_create，若干次 epoll_ctl，一次 epoll_wait，一次 close 等系統調用構成，實際上 epoll 將 poll 分成若干部分實現的原因正是因為服務器軟件中使用 poll 的特點（比如Web服務器）：

1、需要同時 poll 大量文件描述符;

2、每次 poll 完成后就緒的文件描述符只占所有被 poll 的描述符的很少一部分。

3、前后多次 poll 調用對文件描述符數組（ufds）的修改只是很小;

傳統的 poll 函數相當于每次調用都重起爐灶，從用戶空間完整讀入 ufds，完成后再次完全拷貝到用戶空間，另外每次 poll 都需要對所有設備做至少做一次加入和刪除等待隊列操作，這些都是低效的原因。

epoll 將以上情況都細化考慮，不需要每次都完整讀入輸出 ufds，只需使用 epoll_ctl 調整其中一小部分，不需要每次 epoll_wait 都執行一次加入刪除等待隊列操作，另外改進后的機制使的不必在某個設備就緒后搜索整個設備數組進行查找，這些都能提高效率。另外最明顯的一點，從用戶的使用來說，使用 epoll 不必每次都輪詢所有返回結果已找出其中的就緒部分，O(n) 變 O(1)，對性能也提高不少。

此外這里還發現一點，是不是將 epoll_ctl 改成一次可以處理多個 fd（像 semctl 那樣）會提高些許性能呢？特別是在假設系統調用比較耗時的基礎上。不過關于系統調用的耗時問題還會在以后分析。

> POLL/EPOLL 測試數據對比：

測試的環境：我寫了三段代碼來分別模擬服務器，活動的客戶端，僵死的客戶端，服務器運行于一個自編譯的標準 2.6.11 內核系統上，硬件為 PIII933，兩個客戶端各自運行在另外的 PC 上，這兩臺PC比服務器的硬件性能要好，主要是保證能輕易讓服務器滿載，三臺機器間使用一個100M交換機連接。

服務器接受并poll所有連接，如果有request到達則回復一個response，然后繼續poll。

活動的客戶端（Active Client）模擬若干并發的活動連接，這些連接不間斷的發送請求接受回復。

僵死的客戶端（zombie）模擬一些只連接但不發送請求的客戶端，其目的只是占用服務器的poll描述符資源。

測試過程：保持10個并發活動連接，不斷的調整僵并發連接數，記錄在不同比例下使用 poll 與 epoll 的性能差別。僵死并發連接數根據比例分別是：0，10，20，40，80，160，320，640，1280，2560，5120，10240。

下圖中橫軸表示僵死并發連接與活動并發連接之比，縱軸表示完成 40000 次請求回復所花費的時間，以秒為單位。紅色線條表示 poll 數據，綠色表示 epoll 數據。可以看出，poll 在所監控的文件描述符數量增加時，其耗時呈線性增長，而 epoll 則維持了一個平穩的狀態，幾乎不受描述符個數影響。



但是要注意的是在監控的所有客戶端都是活動時，poll 的效率會略高于 epoll（主要在原點附近，即僵死并發連接為0時，圖上不易看出來），究竟 epoll 實現比 poll 復雜，監控少量描述符并非它的長處。

> epoll 的優點綜述

1、支持一個進程打開大數目的socket描述符(FD)：select 最不能忍受的是一個進程所打開的 FD 是有一定限制的，由 FD_SETSIZE 設置，在 Linux 中，這個值是 1024。對于那些需要支持的上萬連接數目的網絡服務器來說顯然太少了。這時候你一是可以選擇修改這個宏然后重新編譯內核，不過資料也同時指出這樣會帶來網絡效率的下降，二是可以選擇多進程的解決方案(傳統的 Apache 方案)，不過雖然 linux 上面創建進程的代價比較小，但仍舊是不可忽視的，加上進程間數據同步遠比不上線程間同步的高效，所以也不是一種完美的方案。不過 epoll 則沒有這個限制，它所支持的 FD 上限是最大可以打開文件的數目，這個數字一般遠大于 1024，舉個例子，在 1GB 內存的機器上大約是 10 萬左右，具體數目可以 cat /proc/sys/fs/file-max 察看，一般來說這個數目和系統內存關系很大。

2、IO 效率不隨 FD 數目增加而線性下降：傳統的 select/poll 另一個致命弱點就是當你擁有一個很大的 socket 集合，不過由于網絡延時，任一時間只有部分的 socket 是"活躍"的，但是 select/poll 每次調用都會線性掃描全部的集合，導致效率呈現線性下降。但是 epoll 不存在這個問題，它只會對"活躍"的 socket 進行操作---這是因為在內核實現中 epoll 是根據每個 fd 上面的 callback 函數實現的。那么，只有"活躍"的 socket 才會主動的去調用 callback 函數，其他 idle 狀態 socket 則不會，在這點上，epoll 實現了一個"偽"AIO，因為這時候推動力在 os 內核。在一些 benchmark 中，如果所有的 socket 基本上都是活躍的 --- 比如一個高速LAN環境，epoll 并不比 select/poll 有什么效率，相反，如果過多使用 epoll_ctl，效率相比還有稍微的下降。但是一旦使用 idle connections 模擬 WAN 環境，epoll 的效率就遠在 select/poll 之上了。

3、使用 mmap 加速內核與用戶空間的消息傳遞：這點實際上涉及到 epoll 的具體實現了。無論是 select，poll 還是 epoll 都需要內核把 FD 消息通知給用戶空間，如何避免不必要的內存拷貝就很重要，在這點上，epoll 是通過內核于用戶空間 mmap 同一塊內存實現的。而如果你想我一樣從 2.5 內核就關注 epoll 的話，一定不會忘記手工 mmap 這一步的。

4、內核微調：這一點其實不算 epoll 的優點了，而是整個 linux 平臺的優點。也許你可以懷疑 linux 平臺，但是你無法回避 linux 平臺賦予你微調內核的能力。比如，內核 TCP/IP 協議棧使用內存池管理 sk_buff 結構，那么可以在運行時期動態調整這個內存 pool(skb_head_pool) 的大小 --- 通過 echo XXXX > /proc/sys/net/core/hot_list_length 完成。再比如 listen 函數的第 2 個參數(TCP 完成 3 次握手的數據包隊列長度)，也可以根據你平臺內存大小動態調整。更甚至在一個數據包面數目巨大但同時每個數據包本身大小卻很小的特殊系統上嘗試最新的 NAPI 網卡驅動架構。

> AIO 和 Epoll

epoll 和 aio（這里的aio是指linux 2.6內核后提供的aio api）的區別：

1、aio 是異步非阻塞的。其實是aio是用線程池實現了異步IO。

2、epoll 在這方面的定義上有點復雜，首先 epoll 的 fd 集里面每一個 fd 都是非阻塞的，但是 epoll（包括 select/poll）在調用時阻塞等待 fd 可用，然后 epoll 只是一個異步通知機制，只是在 fd 可用時通知你，并沒有做任何 IO 操作，所以不是傳統的異步。

在這方面，Windows 無疑是前行者，當然 Boost C++ 庫已經實現了 linux 下 aio 的機制，有興趣的朋友可以參考：http://stlchina.huhoo.net/twiki/bin/view.pl/Main/WebHome

> Reactor 和 Proactor

一般地,I/O多路復用機制都依賴于一個事件多路分離器(Event Demultiplexer)。分離器對象可將來自事件源的I/O事件分離出來，并分發到對應的read/write事件處理器(Event Handler)。開發人員預先注冊需要處理的事件及其事件處理器（或回調函數）；事件分離器負責將請求事件傳遞給事件處理器。兩個與事件分離器有關的模式是Reactor和Proactor。Reactor模式采用同步IO，而Proactor采用異步IO。

在Reactor中，事件分離器負責等待文件描述符或socket為讀寫操作準備就緒，然后將就緒事件傳遞給對應的處理器，最后由處理器負責完成實際的讀寫工作。而在Proactor模式中，處理器--或者兼任處理器的事件分離器，只負責發起異步讀寫操作。IO操作本身由操作系統來完成。傳遞給操作系統的參數需要包括用戶定義的數據緩沖區地址和數據大小，操作系統才能從中得到寫出操作所需數據，或寫入從socket讀到的數據。事件分離器捕獲IO操作完成事件，然后將事件傳遞給對應處理器。比如，在windows上，處理器發起一個異步IO操作，再由事件分離器等待IOCompletion事件。典型的異步模式實現，都建立在操作系統支持異步API的基礎之上，我們將這種實現稱為“系統級”異步或“真”異步，因為應用程序完全依賴操作系統執行真正的IO工作。

舉個例子，將有助于理解Reactor與Proactor二者的差異，以讀操作為例（類操作類似）。

在Reactor中實現讀：
 - 注冊讀就緒事件和相應的事件處理器
 - 事件分離器等待事件
 - 事件到來，激活分離器，分離器調用事件對應的處理器。
 - 事件處理器完成實際的讀操作，處理讀到的數據，注冊新的事件，然后返還控制權。

與如下Proactor（真異步）中的讀過程比較：
 - 處理器發起異步讀操作（注意：操作系統必須支持異步IO）。在這種情況下，處理器無視IO就緒事件，它關注的是完成事件。
 - 事件分離器等待操作完成事件
 - 在分離器等待過程中，操作系統利用并行的內核線程執行實際的讀操作，并將結果數據存入用戶自定義緩沖區，最后通知事件分離器讀操作完成。
 - 事件分離器呼喚處理器。
 - 事件處理器處理用戶自定義緩沖區中的數據，然后啟動一個新的異步操作，并將控制權返回事件分離器。

對于不提供異步 IO API 的操作系統來說，這種辦法可以隱藏 Socket API 的交互細節，從而對外暴露一個完整的異步接口。借此，我們就可以進一步構建完全可移植的，平臺無關的，有通用對外接口的解決方案。上述方案已經由Terabit P/L公司實現為 TProactor (ACE compatible Proactor) ：http://www.terabit.com.au/solutions.php。正是因為 linux 對 aio 支持的不完整，所以 ACE_Proactor 框架在 linux 上的表現很差，大部分在 windows 上執行正常的代碼，在 linux 則運行異常，甚至不能編譯通過。這個問題一直困擾著很大多數 ACE 的用戶，現在好了，有一個 TProactor 幫助解決了在 Linux 不完整支持 AIO 的條件下，正常使用（至少是看起來正常）ACE_Proactor。TProactor 有兩個版本：C++ 和 Java 的。C++ 版本采用 ACE 跨平臺底層類開發，為所有平臺提供了通用統一的主動式異步接口。Boost.Asio 庫，也是采取了類似的這種方案來實現統一的 IO 異步接口。

以下是一張 TProactor 架構設計圖，有興趣的朋友可以看看：



到這里，第二部分的內容結束了，相信大家對 Socket 的底層技術原理有了一個更深層次的理解，在下一篇《Socket深度探究4PHP（三）》我將會深入 PHP 源代碼，探究一下 PHP 在 Socket 這部分的一些技術內幕，然后介紹一下目前在這個領域比較活躍的項目（node.js）。

To be continued ...