學習yield《轉》

時間  2019-11-21
標籤 學習 yield 简体版
原文   原文鏈接

##預備知識 ###Generatorphp

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, "\n";  
}  
/* 
 * 1 
 * 2 
 * ... 
 * 1000 
 */  

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

因爲接觸PHP時日尚淺，並未深刻語言實現細節，因此只能根據現象進行猜想，如下是個人一些我的理解： 包含yield關鍵字的函數比較特殊，返回值是一個Generator對象，此時函數內語句還沒有真正執行 Generator對象是Iterator接口實例，能夠經過rewind()、current()、next()、valid()系列接口進行操縱 Generator能夠視爲一種「可中斷」的函數，而yield構成了一系列的「中斷點」 Generator相似於車間生產的流水線，每次須要用產品的時候才從那裏取一個，而後這個流水線就停在那裏等待下一次取操做 ###Coroutine 細心的讀者可能已經發現，截至目前，其實Generator已經實現了Coroutine的關鍵特性：中斷執行、恢復執行。按照《當C/C++後臺開發趕上Coroutine》的思路，藉助「全局變量」一類語言設施進行信息傳遞，實現異步Server應該足夠了。 其實相對於swapcontext族函數，Generator已經前進了一大步，具有了「返回數據」的能力，若是同時具有「發送數據」的能力，就不再必經過那些蹩腳的手法繞路而行了。在PHP裏面，經過Generator的send()接口（注意：再也不是next()接口），能夠完成「發送數據」的任務，從而實現了真正的「雙向通訊」。python

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

做爲C/C++系碼農，發現「可重入」、「雙向通訊」能力以後，貌似沒有更多奢求了，不過PHP仍是比較慷慨，繼續添加了Exception機制，「錯誤處理」機制獲得進一步完善。網絡

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

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

<?php  
  
class 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())."\n");  
        }  
        if (!socket_set_nonblock($this->socket)) {  
            die(socket_strerror(socket_last_error())."\n");  
        }  
        if(!socket_bind($this->socket, "0.0.0.0", 1234)) {  
            die(socket_strerror(socket_last_error())."\n");  
        }  
    }  
  
    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.\n";  
                    continue;  
                }  
                if ($one == $this->socket) {  
                    //echo "[Run]request recvfrom succ. data=$data ip=$ip port=$port\n";  
                    $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\n";  
                    $socket = socket_create(AF_INET, SOCK_DGRAM, SOL_UDP);  
                    if(!$socket) {  
                        //echo socket_strerror(socket_last_error())."\n";  
                        $coroutine->throw(new Exception(socket_strerror(socket_last_error()), socket_last_error()));  
                        continue;  
                    }  
                    if (!socket_set_nonblock($socket)) {  
                        //echo socket_strerror(socket_last_error())."\n";  
                        $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\n";  
                    if (!isset($this->tasks[$one])) {  
                        //echo "no async_task found.\n";  
                    } 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\n";  
    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\n";  
    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()實現異步網絡交互 ###第二版完善 初版遺留問題：socket

異步網絡交互的timeout未實現，僅預留了接口參數 yield new AsyncTask()調用方式不夠天然，略感彆扭async

<?php  
  
class 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())."\n");  
        }  
        if (!socket_set_nonblock($this->socket)) {  
            die(socket_strerror(socket_last_error())."\n");  
        }  
        if(!socket_bind($this->socket, "0.0.0.0", 1234)) {  
            die(socket_strerror(socket_last_error())."\n");  
        }  
    }  
  
    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.\n";  
                    continue;  
                }  
                if ($one == $this->socket) {  
                    //echo "[Run]request recvfrom succ. data=$data ip=$ip port=$port\n";  
                    $handler = $this->handler;  
                    $coroutine = $handler($one, $data, $len, $ip, $port);  
                    if (!$coroutine) {  
                        //echo "[Run]everything is done.\n";  
                        continue;  
                    }  
                    $task = $coroutine->current();  
                    //echo "[Run]AsyncTask recv. data=$task->data ip=$task->ip port=$task->port timeout=$task->timeout\n";  
                    $socket = socket_create(AF_INET, SOCK_DGRAM, SOL_UDP);  
                    if(!$socket) {  
                        //echo socket_strerror(socket_last_error())."\n";  
                        $coroutine->throw(new Exception(socket_strerror(socket_last_error()), socket_last_error()));  
                        continue;  
                    }  
                    if (!socket_set_nonblock($socket)) {  
                        //echo socket_strerror(socket_last_error())."\n";  
                        $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\n";  
                    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\n";  
    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\n";  
    }  
    //echo "[RequestHandler] after yield AsyncTask. RSP=$rsp_buf\n";  
    socket_sendto($socket, $rsp_buf, $rsp_len, 0, $ip, $port);  
}  
  
$server = new AsyncServer(RequestHandler);  
$server->Run();  
  
?>

代碼解讀：函數

藉助PHP內置array能力，實現簡單的「超時管理」，以毫秒爲精度做爲時間分片 封裝AsyncSendRecv接口，調用形如yield AsyncSendRecv()，更加天然 添加Exception做爲錯誤處理機制，添加ret_code亦可，僅爲展現之用 ##性能測試性能

輸入圖片說明

100Byte/REQ 1000Byte/REQ async_svr_v1.php 16000/s 15000/s async_svr_v2.php 11000/s 10000/s ##展望將來 有興趣的PHPer能夠基於該思路進行底層框架封裝，對於常見阻塞操做進行封裝，好比：connect、send、recv、sleep ... 本人接觸PHP時日尚淺，不少用法非最優，高手可有針對性優化，性能應該能夠繼續提升 目前基於socket進行coroutine綁定，若是基於TCP通訊，每次connect/close，開銷過大，須要考慮實現鏈接池 python等語言也有相似的語言設施，有興趣的讀者能夠自行研究測試

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程