PHP的多任务协程处理的分析
那么,开始吧!
这就是本文我们要讨论的问题。不过我们会从更简单更熟悉的示例开始。
一切从数组开始
我们可以通过简单的遍历来使用数组:
$array = ["foo", "bar", "baz"]; foreach ($array as $key => $value) { print "item: " . $key . "|" . $value . "\n"; } for ($i = 0; $i < count($array); $i++) { print "item: " . $i . "|" . $array[$i] . "\n"; }
这是我们日常编码所依赖的基本实现。可以通过遍历数组获取每个元素的键名和键值。
当然,如果我们希望能够知道在何时可以使用数组。PHP 提供了一个方便的内置函数:
print is_array($array) ? "yes" : "no"; // yes
类数组处理
有时,我们需要对一些数据使用相同的方式进行遍历处理,但它们并非数组类型。比如对 DOMDocument 类进行处理:
$document = new DOMDocument(); $document->loadXML("<p></p>"); $elements = $document->getElementsByTagName("p"); print_r($elements); // DOMNodeList Object ( [length] => 1 )
这显然不是一个数组,但是它有一个 length 属性。我们能像遍历数组一样,对其进行遍历么?我们可以判断它是否实现了下面这个特殊的接口:
print ($elements instanceof Traversable) ? "yes" : "no"; // yes
这真的太有用了。它不会导致我们在遍历非可遍历数据时触发错误。我们仅需在处理前进行检测即可。
不过,这会引发另外一个问题:我们能否让自定义类也拥有这个功能呢?回答是肯定的!第一个实现方法类似如下:
class MyTraversable implements Traversable { // 在这里编码... }
如果我们执行这个类,我们将看到一个错误信息:
PHP Fatal error: Class MyTraversable must implement interface Traversable as part of either Iterator or IteratorAggregate
Iterator(迭代器)
我们无法直接实现 Traversable,但是我们可以尝试第二种方案:
class MyTraversable implements Iterator { // 在这里编码... }
这个接口需要我们实现 5 个方法。让我们完善我们的迭代器:
class MyTraversable implements Iterator { protected $data; protected $index = 0; public function __construct($data) { $this->data = $data; } public function current() { return $this->data[$this->index]; } public function next() { return $this->data[$this->index++]; } public function key() { return $this->index; } public function rewind() { $this->index = 0; } public function valid() { return $this->index < count($this->data); } }
这边我们需要注意几个事项:
我们需要存储构造器方法传入的 $data 数组,以便后续我们可以从中获取它的元素。
还需要一个内部索引(或指针)来跟踪 current 或 next 元素。
rewind() 仅仅重置 index 属性,这样 current() 和 next() 才能正常工作。
键名并非只能是数字类型!这里使用数组索引是为了保证示例足够简单。
我们可以向下面这样运行这段代码:
$iterator = new MyIterator(["foo", "bar", "baz"]); foreach ($iterator as $key => $value) { print "item: " . $key . "|" . $value . "\n"; }
这看起来需要处理太多工作,但是这是能够像数组一样使用 foreach/for 功能的一个简洁实现。
IteratorAggregate(聚合迭代器)
还记得第二个接口抛出的 Traversable 异常么?下面看一个比实现 Iterator 接口更快的实现吧:
class MyIteratorAggregate implements IteratorAggregate { protected $data; public function __construct($data) { $this->data = $data; } public function getIterator() { return new ArrayIterator($this->data); } }
这里我们作弊了。相比于实现一个完整的 Iterator,我们通过 ArrayIterator() 装饰。不过,这相比于通过实现完整的 Iterator 简化了不少代码。
兄弟莫急!先让我们比较一些代码。首先,我们在不使用生成器的情况下从文件中读取每一行数据:
$content = file_get_contents(__FILE__); $lines = explode("\n", $content); foreach ($lines as $i => $line) { print $i . ". " . $line . "\n"; }
这段代码读取文件自身,然后会打印出每行的行号和代码。那么为什么我们不使用生成器呢!
function lines($file) { $handle = fopen($file, 'r'); while (!feof($handle)) { yield trim(fgets($handle)); } fclose($handle); } foreach (lines(__FILE__) as $i => $line) { print $i . ". " . $line . "\n"; }
我知道这看起来更加复杂。不错,不过这是因为我们没有使用 file_get_contents() 函数。一个生成器看起来就像是一个函数,但是它会在每次获取到 yield 关键词是停止运行。
生成器看起来有点像迭代器:
print_r(lines(__FILE__)); // Generator Object ( )
尽管它不是迭代器,它是一个 Generator。它的内部定义了什么方法呢?
print_r(get_class_methods(lines(__FILE__))); // Array // ( // [0] => rewind // [1] => valid // [2] => current // [3] => key // [4] => next // [5] => send // [6] => throw // [7] => __wakeup // )
如果你读取一个大文件,然后使用 memory_get_peak_usage(),你会注意到生成器的代码会使用固定的内存,无论这个文件有多大。它每次进度去一行。而是用 file_get_contents() 函数读取整个文件,会使用更大的内存。这就是在迭代处理这类事物时,生成器的能给我们带来的优势!
Send(发送数据)
可以将数据发送到生成器中。看下下面这个生成器:
<?php $generator = call_user_func(function() { yield "foo"; }); print $generator->current() . "\n"; // foo
注意这里我们如何在 call_user_func() 函数中封装生成器函数的?这里仅仅是一个简单的函数定义,然后立即调用它获取一个新的生成器实例...
我们已经见过 yield 的用法。我们可以通过扩展这个生成器来接收数据:
$generator = call_user_func(function() { $input = (yield "foo"); print "inside: " . $input . "\n"; }); print $generator->current() . "\n"; $generator->send("bar");
数据通过 yield 关键字传入和返回。首先,执行 current() 代码直到遇到 yield,返回 foo。send() 将输出传入到生成器打印输入的位置。你需要习惯这种用法。
抛出异常(Throw)
由于我们需要同这些函数进行交互,可能希望将异常推送到生成器中。这样这些函数就可以自行处理异常。
看看下面这个示例:
$multiply = function($x, $y) { yield $x * $y; }; print $multiply(5, 6)->current(); // 30
现在让我们将它封装到另一个函数中:
$calculate = function ($op, $x, $y) use ($multiply) { if ($op === 'multiply') { $generator = $multiply($x, $y); return $generator->current(); } }; print $calculate("multiply", 5, 6); // 30
这里我们通过一个普通闭包将乘法生成器封装起来。现在让我们验证无效参数:
$calculate = function ($op, $x, $y) use ($multiply) { if ($op === "multiply") { $generator = $multiply($x, $y); if (!is_numeric($x) || !is_numeric($y)) { throw new InvalidArgumentException(); } return $generator->current(); } }; print $calculate('multiply', 5, 'foo'); // PHP Fatal error...
如果我们希望能够通过生成器处理异常?我们怎样才能将异常传入生成器呢!
$multiply = function ($x, $y) { try { yield $x * $y; } catch (InvalidArgumentException $exception) { print "ERRORS!"; } }; $calculate = function ($op, $x, $y) use ($multiply) { if ($op === "multiply") { $generator = $multiply($x, $y); if (!is_numeric($x) || !is_numeric($y)) { $generator->throw(new InvalidArgumentException()); } return $generator->current(); } }; print $calculate('multiply', 5, 'foo'); // PHP Fatal error...
棒呆了!我们不仅可以像迭代器一样使用生成器。还可以通过它们发送数据并抛出异常。它们是可中断和可恢复的函数。有些语言把这些函数叫做……
我们可以使用协程(coroutines)来构建异步代码。让我们来创建一个简单的任务调度程序。首先我们需要一个 Task 类:
class Task { protected $generator; public function __construct(Generator $generator) { $this->generator = $generator; } public function run() { $this->generator->next(); } public function finished() { return !$this->generator->valid(); } }
Task 是普通生成器的装饰器。我们将生成器赋值给它的成员变量以供后续使用,然后实现一个简单的 run() 和 finished() 方法。run() 方法用于执行任务,finished() 方法用于让调度程序知道何时终止运行。
然后我们需要一个 Scheduler 类:
class Scheduler { protected $queue; public function __construct() { $this->queue = new SplQueue(); } public function enqueue(Task $task) { $this->queue->enqueue($task); } pulic function run() { while (!$this->queue->isEmpty()) { $task = $this->queue->dequeue(); $task->run(); if (!$task->finished()) { $this->queue->enqueue($task); } } } }
Scheduler 用于维护一个待执行的任务队列。run() 会弹出队列中的所有任务并执行它,直到运行完整个队列任务。如果某个任务没有执行完毕,当这个任务本次运行完成后,我们将再次入列。
SplQueue 对于这个示例来讲再合适不过了。它是一种 FIFO(先进先出:fist in first out) 数据结构,能够确保每个任务都能够获取足够的处理时间。
我们可以像这样运行这段代码:
$scheduler = new Scheduler(); $task1 = new Task(call_user_func(function() { for ($i = 0; $i < 3; $i++) { print "task1: " . $i . "\n"; yield; } })); $task2 = new Task(call_user_func(function() { for ($i = 0; $i < 6; $i++) { print "task2: " . $i . "\n"; yield; } })); $scheduler->enqueue($task1); $scheduler->enqueue($task2); $scheduler->run();
运行时,我们将看到如下执行结果:
task 1: 0 task 1: 1 task 2: 0 task 2: 1 task 1: 2 task 2: 2 task 2: 3 task 2: 4 task 2: 5
这几乎就是我们想要的执行结果。不过有个问题发生在首次运行每个任务时,它们都执行了两次。我们可以对 Task 类稍作修改来修复这个问题:
class Task { protected $generator; protected $run = false; public function __construct(Generator $generator) { $this->generator = $generator; } public function run() { if ($this->run) { $this->generator->next(); } else { $this->generator->current(); } $this->run = true; } public function finished() { return !$this->generator->valid(); } }
我们需要调整首次 run() 方法调用,从生成器当前有效的指针读取运行。后续调用可以从下一个指针读取运行...
有些人基于这个思路实现了一些超赞的类库。我们来看看其中的两个...
RecoilPHP
RecoilPHP 是一套基于协程的类库,它最令人印象深刻的是用于 ReactPHP 内核。可以将事件循环在 RecoilPHP 和 RecoilPHP 之间进行交换,而你的程序无需架构上的调整。
我们来看一下 ReactPHP 异步 DNS 解决方案:
function resolve($domain, $resolver) { $resolver ->resolve($domain) ->then(function ($ip) use ($domain) { print "domain: " . $domain . "\n"; print "ip: " . $ip . "\n"; }, function ($error) { print $error . "\n"; }) } function run() { $loop = React\EventLoop\Factory::create(); $factory = new React\Dns\Resolver\Factory(); $resolver = $factory->create("8.8.8.8", $loop); resolve("silverstripe.org", $resolver); resolve("wordpress.org", $resolver); resolve("wardrobecms.com", $resolver); resolve("pagekit.com", $resolver); $loop->run(); } run();
resolve() 接收域名和 DNS 解析器,并使用 ReactPHP 执行标准的 DNS 查找。不用太过纠结与 resolve() 函数内部。重要的是这个函数不是生成器,而是一个函数!
run() 创建一个 ReactPHP 事件循环,DNS 解析器(这里是个工厂实例)解析若干域名。同样,这个也不是一个生成器。
想知道 RecoilPHP 到底有何不同?还希望掌握更多细节!
use Recoil\Recoil; function resolve($domain, $resolver) { try { $ip = (yield $resolver->resolve($domain)); print "domain: " . $domain . "\n"; print "ip: " . $ip . "\n"; } catch (Exception $exception) { print $exception->getMessage() . "\n"; } } function run() { $loop = (yield Recoil::eventLoop()); $factory = new React\Dns\Resolver\Factory(); $resolver = $factory->create("8.8.8.8", $loop); yield [ resolve("silverstripe.org", $resolver), resolve("wordpress.org", $resolver), resolve("wardrobecms.com", $resolver), resolve("pagekit.com", $resolver), ]; } Recoil::run("run");
通过将它集成到 ReactPHP 来完成一些令人称奇的工作。每次运行 resolve() 时,RecoilPHP 会管理由 $resoler->resolve() 返回的 promise 对象,然后将数据发送给生成器。此时我们就像在编写同步代码一样。与我们在其他一步模型中使用回调代码不同,这里只有一个指令列表。
RecoilPHP 知道它应该管理一个有执行 run() 函数时返回的 yield 数组。RoceilPHP 还支持基于协程的数据库(PDO)和日志库。
IcicleIO
IcicleIO 为了一全新的方案实现 ReactPHP 一样的目标,而仅仅使用协程功能。相比 ReactPHP 它仅包含极少的组件。但是,核心的异步流、服务器、Socket、事件循环特性一个不落。
让我们看一个 socket 服务器示例:
use Icicle\Coroutine\Coroutine; use Icicle\Loop\Loop; use Icicle\Socket\Client\ClientInterface; use Icicle\Socket\Server\ServerInterface; use Icicle\Socket\Server\ServerFactory; $factory = new ServerFactory(); $coroutine = Coroutine::call(function (ServerInterface $server) { $clients = new SplObjectStorage(); $handler = Coroutine::async( function (ClientInterface $client) use (&$clients) { $clients->attach($client); $host = $client->getRemoteAddress(); $port = $client->getRemotePort(); $name = $host . ":" . $port; try { foreach ($clients as $stream) { if ($client !== $stream) { $stream->write($name . "connected.\n"); } } yield $client->write("Welcome " . $name . "!\n"); while ($client->isReadable()) { $data = trim(yield $client->read()); if ("/exit" === $data) { yield $client->end("Goodbye!\n"); } else { $message = $name . ":" . $data . "\n"; foreach ($clients as $stream) { if ($client !== $stream) { $stream->write($message); } } } } } catch (Exception $exception) { $client->close($exception); } finally { $clients->detach($client); foreach ($clients as $stream) { $stream->write($name . "disconnected.\n"); } } } ); while ($server->isOpen()) { $handler(yield $server->accept()); } }, $factory->create("127.0.0.1", 6000)); Loop::run();
据我所知,这段代码所做的事情如下:
在 127.0.0.1 和 6000 端口创建一个服务器实例,然后将其传入外部生成器.
外部生成器运行,同时服务器等待新连接。当服务器接收一个连接它将其传入内部生成器。
内部生成器写入消息到 socket。当 socket 可读时运行。
每次 socket 向服务器发送消息时,内部生成器检测消息是否是退出标识。如果是,通知其他 socket。否则,其它 socket 发送这个相同的消息。
打开命令行终端输入 nc localhost 6000 查看执行结果!
该示例使用 SplObjectStorage 跟踪 socket 连接。这样我们就可以向所有 socket 发送消息。
这个话题可以包含很多内容。希望您能看到生成器是如何创建的,以及它们如何帮助编写迭代程序和异步代码。
如果你有问题,可以随时问我。
相关推荐:
以上就是PHP的多任务协程处理的分析的详细内容,更多请关注其它相关文章!