By warezhou 2014.11.24
上次通過C擴展為PHP添加coroutine嘗試失敗之后，由于短期內啃下Zend可能性幾乎為零，只能打語言原生能力的主意了。Google之后發現，PHP5.5引入了Generator和Coroutine新特性，于是才有了本文的誕生。
預備知識Generatorfunction 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時日尚淺，并未深入語言實現細節，所以只能根據現象進行猜測，以下是我的一些個人理解：
包含yield關鍵字的函數比較特殊，返回值是一個Generator對象，此時函數內語句尚未真正執行Generator對象是Iterator接口實例，可以通過rewind()、html' target='_blank'>current()、next()、valid()系列接口進行操縱Generator可以視為一種“可中斷”的函數，而yield構成了一系列的“中斷點”Generator類似于車間生產的流水線，每次需要用產品的時候才從那里取一個，然后這個流水線就停在那里等待下一次取操作Coroutine細心的讀者可能已經發現，截至目前，其實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();?>
代碼解讀：
為了便于說明問題，這里所有底層通訊基于UDP，省略了TCP的connect等繁瑣細節AsyncServer為底層框架類，封裝了網絡通訊細節以及協程切換細節，通過socket進行coroutine綁定RequestHandler為業務處理函數，通過yield new AsyncTask()實現異步網絡交互第二版完善第一版遺留問題：
異步網絡交互的timeout未實現，僅預留了接口參數yield new AsyncTask()調用方式不夠自然，略感別扭<?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();?>代碼解讀：
借助PHP內置array能力，實現簡單的“超時管理”，以毫秒為精度作為時間分片封裝AsyncSendRecv接口，調用形如yield AsyncSendRecv()，更加自然添加Exception作為錯誤處理機制，添加ret_code亦可，僅為展示之用性能測試測試環境
鄭重聲明：本文版權歸原作者所有，轉載文章僅為傳播更多信息之目的，如作者信息標記有誤，請第一時間聯系我們修改或刪除，多謝。
新聞熱點
疑難解答
