PHP的多任务协程处理的分析

那么，开始吧！

这就是本文我们要讨论的问题。不过我们会从更简单更熟悉的示例开始。

我们可以通过简单的遍历来使用数组：

$array = [ foo , bar , baz foreach ($array as $key = $html' target='_blank'>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 // 在这里编码...}

如果我们执行这个类，我们将看到一个错误信息：

我们无法直接实现 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 功能的一个简洁实现。

还记得第二个接口抛出的 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// )

可以将数据发送到生成器中。看下下面这个生成器：

 ?php$generator = call_user_func(function() { yield foo print $generator- current() . /n // foo

我们已经见过 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() 将输出传入到生成器打印输入的位置。你需要习惯这种用法。

由于我们需要同这些函数进行交互，可能希望将异常推送到生成器中。这样这些函数就可以自行处理异常。

看看下面这个示例：

$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() 会弹出队列中的所有任务并执行它，直到运行完整个队列任务。如果某个任务没有执行完毕，当这个任务本次运行完成后，我们将再次入列。

我们可以像这样运行这段代码：

$scheduler = new Scheduler();$task1 = new Task(call_user_func(function() { for ($i = 0; $i $i++) { print task1: . $i . /n  yield;$task2 = new Task(call_user_func(function() { for ($i = 0; $i $i++) { print task2: . $i . /n  yield;$scheduler- enqueue($task1);$scheduler- enqueue($task2);$scheduler- run();

运行时，我们将看到如下执行结果：

task 1: 0task 1: 1task 2: 0task 2: 1task 1: 2task 2: 2task 2: 3task 2: 4task 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 是一套基于协程的类库，它最令人印象深刻的是用于 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 为了一全新的方案实现 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 查看执行结果！

这个话题可以包含很多内容。希望您能看到生成器是如何创建的，以及它们如何帮助编写迭代程序和异步代码。

如果你有问题，可以随时问我。

相关推荐：

浅谈一下PHP生成器的使用方法

php实现多进程 多任务的一例代码

以上就是PHP的多任务协程处理的分析的详细内容，PHP教程

郑重声明：本文版权归原作者所有，转载文章仅为传播更多信息之目的，如作者信息标记有误，请第一时间联系我们修改或删除，多谢。