[转]php的协程体验

原文： https://blog.csdn.net/cszhouwei/article/details/41446687?utm_medium=distribute.pc_aggpage_search_result.none-task-blog-2~all~first_rank_v2~rank_v28-12-41446687.nonecase&utm_term=c%E5%8D%8F%E7%A8%8B%E4%BD%BF%E7%94%A8%E5%9C%BA%E6%99%AF&spm=1000.2123.3001.4430

————————————————————————————————————————————————————————

上次通过C扩展为PHP添加coroutine尝试失败之后，由于短期内啃下Zend可能性几乎为零，只能打语言原生能力的主意了。Google之后发现，PHP5.5引入了Generator和Coroutine新特性，于是才有了本文的诞生。

背景阅读

《当C/C++后台开发遇上Coroutine》

http://blog.csdn.net/cszhouwei/article/details/14230529

《一次失败的PHP扩展开发之旅》

http://blog.csdn.net/cszhouwei/article/details/41290673

预备知识

Generator

1.  
   function my_range($start, $end, $step = 1) {
2.  
   for ($i = $start; $i <= $end; $i += $step) {
3.  
   yield $i;
4.  
   }
5.  
   }
6.  
    
7.  
   foreach (my_range(1, 1000) as $num) {
8.  
   echo $num, " ";
9.  
   }
10.  
    /*
11.  
    * 1
12.  
    * 2
13.  
    * ...
14.  
    * 1000
15.  
    */

图 1 基于generator的range()实现

1.  
   $range = my_range(1, 1000);
2.  
    
3.  
   var_dump($range);
4.  
   /*
5.  
   * object(Generator)#1 (0) {
6.  
   * }
7.  
   */
8.  
    
9.  
   var_dump($range instanceof Iterator);
10.  
    /*
11.  
    * bool(true)
12.  
    */

图 2 my_range()的实现推测

由于接触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()接口），可以完成“发送数据”的任务，从而实现了真正的“双向通信”。

1.  
   function gen() {
2.  
   $ret = (yield 'yield1');
3.  
   echo "[gen]", $ret, " ";
4.  
   $ret = (yield 'yield2');
5.  
   echo "[gen]", $ret, " ";
6.  
   }
7.  
    
8.  
   $gen = gen();
9.  
   $ret = $gen->current();
10.  
    echo "[main]", $ret, " ";
11.  
    $ret = $gen->send("send1");
12.  
    echo "[main]", $ret, " ";
13.  
    $ret = $gen->send("send2");
14.  
    echo "[main]", $ret, " ";
15.  
     
16.  
    /*
17.  
    * [main]yield1
18.  
    * [gen]send1
19.  
    * [main]yield2
20.  
    * [gen]send2
21.  
    * [main]
22.  
    */

图 3 Coroutine双向通信示例

作为C/C++系码农，发现“可重入”、“双向通信”能力之后，貌似没有更多奢求了，不过PHP还是比较慷慨，继续添加了Exception机制，“错误处理”机制得到进一步完善。

1.  
   function gen() {
2.  
   $ret = (yield 'yield1');
3.  
   echo "[gen]", $ret, " ";
4.  
   try {
5.  
   $ret = (yield 'yield2');
6.  
   echo "[gen]", $ret, " ";
7.  
   } catch (Exception $ex) {
8.  
   echo "[gen][Exception]", $ex->getMessage(), " ";
9.  
   }
10.  
    echo "[gen]finish ";
11.  
    }
12.  
     
13.  
    $gen = gen();
14.  
    $ret = $gen->current();
15.  
    echo "[main]", $ret, " ";
16.  
    $ret = $gen->send("send1");
17.  
    echo "[main]", $ret, " ";
18.  
    $ret = $gen->throw(new Exception("Test"));
19.  
    echo "[main]", $ret, " ";
20.  
     
21.  
    /*
22.  
    * [main]yield1
23.  
    * [gen]send1
24.  
    * [main]yield2
25.  
    * [gen][Exception]Test
26.  
    * [gen]finish
27.  
    * [main]
28.  
    */

图 4 Coroutine错误处理示例

实战演习

前面简单介绍了相关的语言设施，那么具体到实际项目中，到底应该如何运用呢？让我们继续《一次失败的PHP扩展开发之旅》描述的场景，借助上述特性实现那个美好的愿望：以同步方式书写异步代码！

第一版初稿

1.  
   <?php
2.  
    
3.  
   class AsyncServer {
4.  
   protected $handler;
5.  
   protected $socket;
6.  
   protected $tasks = [];
7.  
    
8.  
   public function __construct($handler) {
9.  
   $this->handler = $handler;
10.  
     
11.  
    $this->socket = socket_create(AF_INET, SOCK_DGRAM, SOL_UDP);
12.  
    if(!$this->socket) {
13.  
    die(socket_strerror(socket_last_error())." ");
14.  
    }
15.  
    if (!socket_set_nonblock($this->socket)) {
16.  
    die(socket_strerror(socket_last_error())." ");
17.  
    }
18.  
    if(!socket_bind($this->socket, "0.0.0.0", 1234)) {
19.  
    die(socket_strerror(socket_last_error())." ");
20.  
    }
21.  
    }
22.  
     
23.  
    public function Run() {
24.  
    while (true) {
25.  
    $reads = array($this->socket);
26.  
    foreach ($this->tasks as list($socket)) {
27.  
    $reads[] = $socket;
28.  
    }
29.  
    $writes = NULL;
30.  
    $excepts= NULL;
31.  
    if (!socket_select($reads, $writes, $excepts, 0, 1000)) {
32.  
    continue;
33.  
    }
34.  
     
35.  
    foreach ($reads as $one) {
36.  
    $len = socket_recvfrom($one, $data, 65535, 0, $ip, $port);
37.  
    if (!$len) {
38.  
    //echo "socket_recvfrom fail. ";
39.  
    continue;
40.  
    }
41.  
    if ($one == $this->socket) {
42.  
    //echo "[Run]request recvfrom succ. data=$data ip=$ip port=$port ";
43.  
    $handler = $this->handler;
44.  
    $coroutine = $handler($one, $data, $len, $ip, $port);
45.  
    $task = $coroutine->current();
46.  
    //echo "[Run]AsyncTask recv. data=$task->data ip=$task->ip port=$task->port timeout=$task->timeout ";
47.  
    $socket = socket_create(AF_INET, SOCK_DGRAM, SOL_UDP);
48.  
    if(!$socket) {
49.  
    //echo socket_strerror(socket_last_error())." ";
50.  
    $coroutine->throw(new Exception(socket_strerror(socket_last_error()), socket_last_error()));
51.  
    continue;
52.  
    }
53.  
    if (!socket_set_nonblock($socket)) {
54.  
    //echo socket_strerror(socket_last_error())." ";
55.  
    $coroutine->throw(new Exception(socket_strerror(socket_last_error()), socket_last_error()));
56.  
    continue;
57.  
    }
58.  
    socket_sendto($socket, $task->data, $task->len, 0, $task->ip, $task->port);
59.  
    $this->tasks[$socket] = [$socket, $coroutine];
60.  
    } else {
61.  
    //echo "[Run]response recvfrom succ. data=$data ip=$ip port=$port ";
62.  
    if (!isset($this->tasks[$one])) {
63.  
    //echo "no async_task found. ";
64.  
    } else {
65.  
    list($socket, $coroutine) = $this->tasks[$one];
66.  
    unset($this->tasks[$one]);
67.  
    socket_close($socket);
68.  
    $coroutine->send(array($data, $len));
69.  
    }
70.  
    }
71.  
    }
72.  
    }
73.  
    }
74.  
    }
75.  
     
76.  
    class AsyncTask {
77.  
    public $data;
78.  
    public $len;
79.  
    public $ip;
80.  
    public $port;
81.  
    public $timeout;
82.  
     
83.  
    public function __construct($data, $len, $ip, $port, $timeout) {
84.  
    $this->data = $data;
85.  
    $this->len = $len;
86.  
    $this->ip = $ip;
87.  
    $this->port = $port;
88.  
    $this->timeout = $timeout;
89.  
    }
90.  
    }
91.  
     
92.  
    function RequestHandler($socket, $req_buf, $req_len, $ip, $port) {
93.  
    //echo "[RequestHandler] before yield AsyncTask. REQ=$req_buf ";
94.  
    list($rsp_buf, $rsp_len) = (yield new AsyncTask($req_buf, $req_len, "127.0.0.1", 2345, 1000));
95.  
    //echo "[RequestHandler] after yield AsyncTask. RSP=$rsp_buf ";
96.  
    socket_sendto($socket, $rsp_buf, $rsp_len, 0, $ip, $port);
97.  
    }
98.  
     
99.  
    $server = new AsyncServer(RequestHandler);
100.  
     $server->Run();
101.  
      
102.  
     ?>

代码解读：

• 为了便于说明问题，这里所有底层通讯基于UDP，省略了TCP的connect等繁琐细节
• AsyncServer为底层框架类，封装了网络通讯细节以及协程切换细节，通过socket进行coroutine绑定
• RequestHandler为业务处理函数，通过yield new AsyncTask()实现异步网络交互

第二版完善

第一版遗留问题：

• 异步网络交互的timeout未实现，仅预留了接口参数
• yield new AsyncTask()调用方式不够自然，略感别扭

1.  
   <?php
2.  
    
3.  
   class AsyncServer {
4.  
   protected $handler;
5.  
   protected $socket;
6.  
   protected $tasks = [];
7.  
   protected $timers = [];
8.  
    
9.  
   public function __construct(callable $handler) {
10.  
    $this->handler = $handler;
11.  
     
12.  
    $this->socket = socket_create(AF_INET, SOCK_DGRAM, SOL_UDP);
13.  
    if(!$this->socket) {
14.  
    die(socket_strerror(socket_last_error())." ");
15.  
    }
16.  
    if (!socket_set_nonblock($this->socket)) {
17.  
    die(socket_strerror(socket_last_error())." ");
18.  
    }
19.  
    if(!socket_bind($this->socket, "0.0.0.0", 1234)) {
20.  
    die(socket_strerror(socket_last_error())." ");
21.  
    }
22.  
    }
23.  
     
24.  
    public function Run() {
25.  
    while (true) {
26.  
    $now = microtime(true) * 1000;
27.  
    foreach ($this->timers as $time => $sockets) {
28.  
    if ($time > $now) break;
29.  
    foreach ($sockets as $one) {
30.  
    list($socket, $coroutine) = $this->tasks[$one];
31.  
    unset($this->tasks[$one]);
32.  
    socket_close($socket);
33.  
    $coroutine->throw(new Exception("Timeout"));
34.  
    }
35.  
    unset($this->timers[$time]);
36.  
    }
37.  
     
38.  
    $reads = array($this->socket);
39.  
    foreach ($this->tasks as list($socket)) {
40.  
    $reads[] = $socket;
41.  
    }
42.  
    $writes = NULL;
43.  
    $excepts= NULL;
44.  
    if (!socket_select($reads, $writes, $excepts, 0, 1000)) {
45.  
    continue;
46.  
    }
47.  
     
48.  
    foreach ($reads as $one) {
49.  
    $len = socket_recvfrom($one, $data, 65535, 0, $ip, $port);
50.  
    if (!$len) {
51.  
    //echo "socket_recvfrom fail. ";
52.  
    continue;
53.  
    }
54.  
    if ($one == $this->socket) {
55.  
    //echo "[Run]request recvfrom succ. data=$data ip=$ip port=$port ";
56.  
    $handler = $this->handler;
57.  
    $coroutine = $handler($one, $data, $len, $ip, $port);
58.  
    if (!$coroutine) {
59.  
    //echo "[Run]everything is done. ";
60.  
    continue;
61.  
    }
62.  
    $task = $coroutine->current();
63.  
    //echo "[Run]AsyncTask recv. data=$task->data ip=$task->ip port=$task->port timeout=$task->timeout ";
64.  
    $socket = socket_create(AF_INET, SOCK_DGRAM, SOL_UDP);
65.  
    if(!$socket) {
66.  
    //echo socket_strerror(socket_last_error())." ";
67.  
    $coroutine->throw(new Exception(socket_strerror(socket_last_error()), socket_last_error()));
68.  
    continue;
69.  
    }
70.  
    if (!socket_set_nonblock($socket)) {
71.  
    //echo socket_strerror(socket_last_error())." ";
72.  
    $coroutine->throw(new Exception(socket_strerror(socket_last_error()), socket_last_error()));
73.  
    continue;
74.  
    }
75.  
    socket_sendto($socket, $task->data, $task->len, 0, $task->ip, $task->port);
76.  
    $deadline = $now + $task->timeout;
77.  
    $this->tasks[$socket] = [$socket, $coroutine, $deadline];
78.  
    $this->timers[$deadline][$socket] = $socket;
79.  
    } else {
80.  
    //echo "[Run]response recvfrom succ. data=$data ip=$ip port=$port ";
81.  
    list($socket, $coroutine, $deadline) = $this->tasks[$one];
82.  
    unset($this->tasks[$one]);
83.  
    unset($this->timers[$deadline][$one]);
84.  
    socket_close($socket);
85.  
    $coroutine->send(array($data, $len));
86.  
    }
87.  
    }
88.  
    }
89.  
    }
90.  
    }
91.  
     
92.  
    class AsyncTask {
93.  
    public $data;
94.  
    public $len;
95.  
    public $ip;
96.  
    public $port;
97.  
    public $timeout;
98.  
     
99.  
    public function __construct($data, $len, $ip, $port, $timeout) {
100.  
     $this->data = $data;
101.  
     $this->len = $len;
102.  
     $this->ip = $ip;
103.  
     $this->port = $port;
104.  
     $this->timeout = $timeout;
105.  
     }
106.  
     }
107.  
      
108.  
     function AsyncSendRecv($req_buf, $req_len, $ip, $port, $timeout) {
109.  
     return new AsyncTask($req_buf, $req_len, $ip, $port, $timeout);
110.  
     }
111.  
      
112.  
     function RequestHandler($socket, $req_buf, $req_len, $ip, $port) {
113.  
     //echo "[RequestHandler] before yield AsyncTask. REQ=$req_buf ";
114.  
     try {
115.  
     list($rsp_buf, $rsp_len) = (yield AsyncSendRecv($req_buf, $req_len, "127.0.0.1", 2345, 3000));
116.  
     } catch (Exception $ex) {
117.  
     $rsp_buf = $ex->getMessage();
118.  
     $rsp_len = strlen($rsp_buf);
119.  
     //echo "[Exception]$rsp_buf ";
120.  
     }
121.  
     //echo "[RequestHandler] after yield AsyncTask. RSP=$rsp_buf ";
122.  
     socket_sendto($socket, $rsp_buf, $rsp_len, 0, $ip, $port);
123.  
     }
124.  
      
125.  
     $server = new AsyncServer(RequestHandler);
126.  
     $server->Run();
127.  
      
128.  
     ?>

代码解读：

• 借助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等语言也有类似的语言设施，有兴趣的读者可以自行研究