学习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等语言也有相似的语言设施，有兴趣的读者能够自行研究测试
相关文章
- 1. python学习——yield和yield from
- 2. python yield学习
- 3. yield学习笔记
- 4. python yield 与 yield from转
- 5. python进阶学习之 yield
- 6. 学习-Pytest(五)yield操做
- 7. yield用法 （转载）
- 8. yield学习笔记（1）——Python yield 使用浅析
- 9. Python 2.5版 yield 之学习心得
- 10. python学习笔记—yield生成器
- 更多相关文章...
- • 您已经学习了 XML Schema，下一步学习什么呢？ - XML Schema 教程
- • 我们已经学习了 SQL，下一步学习什么呢？ - SQL 教程
- • Tomcat学习笔记（史上最全tomcat学习笔记）
- • 适用于PHP初学者的学习线路和建议
相关标签/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
欢迎关注本站公众号,获取更多信息
