By warezhou 2014.11.24

上次通過C擴展為PHP添加coroutine嘗試失敗之后，由于短期內啃下Zend可能性幾乎為零，只能打語言原生能力的主意了。Google之后發現，PHP5.5引入了Generator和Coroutine新特性，于是才有了本文的誕生。

function my_range($start, $end, $step = 1) {    for ($i = $start; $i <= $end; $i += $step) {        yield $i;    }}foreach (my_range(1, 1000) as $num) {    echo $num, "";}/* * 1 * 2 * ... * 1000 */

圖 1 基于generator的range()實現

$range = my_range(1, 1000);var_dump($range);/* * object(Generator)#1 (0) { * } */var_dump($range instanceof Iterator);/* * bool(true) */

圖 2 my_range()的實現推測

由于接觸PHP時日尚淺，并未深入語言實現細節，所以只能根據現象進行猜測，以下是我的一些個人理解：

細心的讀者可能已經發現，截至目前，其實Generator已經實現了Coroutine的關鍵特性：中斷執行、恢復執行。按照《當C/C++后臺開發遇上Coroutine》的思路，借助“全局變量”一類語言設施進行信息傳遞，實現異步Server應該足夠了。

其實相對于swapcontext族函數，Generator已經前進了一大步，具備了“返回數據”的能力，如果同時具備“發送數據”的能力，就再也不必通過那些蹩腳的手法繞路而行了。在PHP里面，通過Generator的send()接口（注意：不再是next()接口），可以完成“發送數據”的任務，從而實現了真正的“雙向通信”。

function gen() {    $ret = (yield 'yield1');    echo "[gen]", $ret, "";    $ret = (yield 'yield2');    echo "[gen]", $ret, "";}$gen = gen();$ret = $gen->current();echo "[main]", $ret, "";$ret = $gen->send("send1");echo "[main]", $ret, "";$ret = $gen->send("send2");echo "[main]", $ret, "";/* * [main]yield1 * [gen]send1 * [main]yield2 * [gen]send2 * [main] */

圖 3 Coroutine雙向通信示例

作為C/C++系碼農，發現“可重入”、“雙向通信”能力之后，貌似沒有更多奢求了，不過PHP還是比較慷慨，繼續添加了Exception機制，“錯誤處理”機制得到進一步完善。

function gen() {    $ret = (yield 'yield1');    echo "[gen]", $ret, "";    try {        $ret = (yield 'yield2');        echo "[gen]", $ret, "";    } catch (Exception $ex) {        echo "[gen][Exception]", $ex->getMessage(), "";    }       echo "[gen]finish";}$gen = gen();$ret = $gen->current();echo "[main]", $ret, "";$ret = $gen->send("send1");echo "[main]", $ret, "";$ret = $gen->throw(new Exception("Test"));echo "[main]", $ret, "";/* * [main]yield1 * [gen]send1 * [main]yield2 * [gen][Exception]Test * [gen]finish * [main] */

圖 4 Coroutine錯誤處理示例

前面簡單介紹了相關的語言設施，那么具體到實際項目中，到底應該如何運用呢？讓我們繼續《一次失敗的PHP擴展開發之旅》描述的場景，借助上述特性實現那個美好的愿望：以同步方式書寫異步代碼！

<?phpclass AsyncServer {    protected $handler;    protected $socket;    protected $tasks = [];    public function __construct($handler) {        $this->handler = $handler;        $this->socket = socket_create(AF_INET, SOCK_DGRAM, SOL_UDP);        if(!$this->socket) {            die(socket_strerror(socket_last_error())."");        }        if (!socket_set_nonblock($this->socket)) {            die(socket_strerror(socket_last_error())."");        }        if(!socket_bind($this->socket, "0.0.0.0", 1234)) {            die(socket_strerror(socket_last_error())."");        }    }    public function Run() {        while (true) {            $reads = array($this->socket);            foreach ($this->tasks as list($socket)) {                $reads[] = $socket;            }            $writes = NULL;            $excepts= NULL;            if (!socket_select($reads, $writes, $excepts, 0, 1000)) {                continue;            }            foreach ($reads as $one) {                $len = socket_recvfrom($one, $data, 65535, 0, $ip, $port);                if (!$len) {                    //echo "socket_recvfrom fail.";                    continue;                }                if ($one == $this->socket) {                    //echo "[Run]request recvfrom succ. data=$data ip=$ip port=$port";                    $handler = $this->handler;                    $coroutine = $handler($one, $data, $len, $ip, $port);                    $task = $coroutine->current();                    //echo "[Run]AsyncTask recv. data=$task->data ip=$task->ip port=$task->port timeout=$task->timeout";                    $socket = socket_create(AF_INET, SOCK_DGRAM, SOL_UDP);                    if(!$socket) {                        //echo socket_strerror(socket_last_error())."";                        $coroutine->throw(new Exception(socket_strerror(socket_last_error()), socket_last_error()));                        continue;                    }                    if (!socket_set_nonblock($socket)) {                        //echo socket_strerror(socket_last_error())."";                        $coroutine->throw(new Exception(socket_strerror(socket_last_error()), socket_last_error()));                        continue;                    }                    socket_sendto($socket, $task->data, $task->len, 0, $task->ip, $task->port);                    $this->tasks[$socket] = [$socket, $coroutine];                } else {                    //echo "[Run]response recvfrom succ. data=$data ip=$ip port=$port";                    if (!isset($this->tasks[$one])) {                        //echo "no async_task found.";                    } else {                        list($socket, $coroutine) = $this->tasks[$one];                        unset($this->tasks[$one]);                        socket_close($socket);                        $coroutine->send(array($data, $len));                    }                }            }        }    }}class AsyncTask {    public $data;    public $len;    public $ip;    public $port;    public $timeout;    public function __construct($data, $len, $ip, $port, $timeout) {        $this->data = $data;        $this->len = $len;        $this->ip = $ip;        $this->port = $port;        $this->timeout = $timeout;    }}function RequestHandler($socket, $req_buf, $req_len, $ip, $port) {    //echo "[RequestHandler] before yield AsyncTask. REQ=$req_buf";    list($rsp_buf, $rsp_len) = (yield new AsyncTask($req_buf, $req_len, "127.0.0.1", 2345, 1000));    //echo "[RequestHandler] after yield AsyncTask. RSP=$rsp_buf";    socket_sendto($socket, $rsp_buf, $rsp_len, 0, $ip, $port);}$server = new AsyncServer(RequestHandler);$server->Run();?>

代碼解讀：

第一版遺留問題：

<?phpclass AsyncServer {    protected $handler;    protected $socket;    protected $tasks = [];    protected $timers = [];    public function __construct(callable $handler) {        $this->handler = $handler;        $this->socket = socket_create(AF_INET, SOCK_DGRAM, SOL_UDP);        if(!$this->socket) {            die(socket_strerror(socket_last_error())."");        }        if (!socket_set_nonblock($this->socket)) {            die(socket_strerror(socket_last_error())."");        }        if(!socket_bind($this->socket, "0.0.0.0", 1234)) {            die(socket_strerror(socket_last_error())."");        }    }    public function Run() {        while (true) {            $now = microtime(true) * 1000;            foreach ($this->timers as $time => $sockets) {                if ($time > $now) break;                foreach ($sockets as $one) {                    list($socket, $coroutine) = $this->tasks[$one];                    unset($this->tasks[$one]);                    socket_close($socket);                    $coroutine->throw(new Exception("Timeout"));                }                unset($this->timers[$time]);            }            $reads = array($this->socket);            foreach ($this->tasks as list($socket)) {                $reads[] = $socket;            }            $writes = NULL;            $excepts= NULL;            if (!socket_select($reads, $writes, $excepts, 0, 1000)) {                continue;            }            foreach ($reads as $one) {                $len = socket_recvfrom($one, $data, 65535, 0, $ip, $port);                if (!$len) {                    //echo "socket_recvfrom fail.";                    continue;                }                if ($one == $this->socket) {                    //echo "[Run]request recvfrom succ. data=$data ip=$ip port=$port";                    $handler = $this->handler;                    $coroutine = $handler($one, $data, $len, $ip, $port);                    if (!$coroutine) {                        //echo "[Run]everything is done.";                        continue;                    }                    $task = $coroutine->current();                    //echo "[Run]AsyncTask recv. data=$task->data ip=$task->ip port=$task->port timeout=$task->timeout";                    $socket = socket_create(AF_INET, SOCK_DGRAM, SOL_UDP);                    if(!$socket) {                        //echo socket_strerror(socket_last_error())."";                        $coroutine->throw(new Exception(socket_strerror(socket_last_error()), socket_last_error()));                        continue;                    }                    if (!socket_set_nonblock($socket)) {                        //echo socket_strerror(socket_last_error())."";                        $coroutine->throw(new Exception(socket_strerror(socket_last_error()), socket_last_error()));                        continue;                    }                    socket_sendto($socket, $task->data, $task->len, 0, $task->ip, $task->port);                    $deadline = $now + $task->timeout;                    $this->tasks[$socket] = [$socket, $coroutine, $deadline];                    $this->timers[$deadline][$socket] = $socket;                } else {                    //echo "[Run]response recvfrom succ. data=$data ip=$ip port=$port";                    list($socket, $coroutine, $deadline) = $this->tasks[$one];                    unset($this->tasks[$one]);                    unset($this->timers[$deadline][$one]);                    socket_close($socket);                    $coroutine->send(array($data, $len));                }            }        }    }}class AsyncTask {    public $data;    public $len;    public $ip;    public $port;    public $timeout;    public function __construct($data, $len, $ip, $port, $timeout) {        $this->data = $data;        $this->len = $len;        $this->ip = $ip;        $this->port = $port;        $this->timeout = $timeout;    }}function AsyncSendRecv($req_buf, $req_len, $ip, $port, $timeout) {    return new AsyncTask($req_buf, $req_len, $ip, $port, $timeout);}function RequestHandler($socket, $req_buf, $req_len, $ip, $port) {    //echo "[RequestHandler] before yield AsyncTask. REQ=$req_buf";    try {        list($rsp_buf, $rsp_len) = (yield AsyncSendRecv($req_buf, $req_len, "127.0.0.1", 2345, 3000));    } catch (Exception $ex) {        $rsp_buf = $ex->getMessage();        $rsp_len = strlen($rsp_buf);        //echo "[Exception]$rsp_buf";    }    //echo "[RequestHandler] after yield AsyncTask. RSP=$rsp_buf";    socket_sendto($socket, $rsp_buf, $rsp_len, 0, $ip, $port);}$server = new AsyncServer(RequestHandler);$server->Run();?>

代碼解讀：

鄭重聲明：本文版權歸原作者所有，轉載文章僅為傳播更多信息之目的，如作者信息標記有誤，請第一時間聯系我們修改或刪除，多謝。