登录  /  注册

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

不言
发布: 2018-07-17 10:20:13
原创
1418人浏览过

这篇文章主要介绍了关于php的多任务协程处理,有着一定的参考价值,现在分享给大家,有需要的朋友可以参考一下

那么,开始吧!

3615091407-5b4d4c9d0f16b_articlex[1].jpg

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

一切从数组开始

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

$array = ["foo", "bar", "baz"];
 
foreach ($array as $key => $value) {
    print "item: " . $key . "|" . $value . "\n";
}
 
for ($i = 0; $i <p>这是我们日常编码所依赖的基本实现。可以通过遍历数组获取每个元素的键名和键值。</p><p>当然,如果我们希望能够知道在何时可以使用数组。PHP 提供了一个方便的内置函数:</p><pre class="brush:php;toolbar:false">print is_array($array) ? "yes" : "no"; // yes
登录后复制

类数组处理

有时,我们需要对一些数据使用相同的方式进行遍历处理,但它们并非数组类型。比如对 DOMDocument 类进行处理:

$document = new DOMDocument();
$document-&gt;loadXML("<p></p>");

$elements = $document-&gt;getElementsByTagName("p");
print_r($elements); // DOMNodeList Object ( [length] =&gt; 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-&gt;data = $data;
    }

    public function current()
    {
        return $this-&gt;data[$this-&gt;index];
    }

    public function next()
    {
        return $this-&gt;data[$this-&gt;index++];
    }

    public function key()
    {
        return $this-&gt;index;
    }

    public function rewind()
    {
        $this-&gt;index = 0;
    }

    public function valid()
    {
        return $this-&gt;index data);
    }
}
登录后复制

这边我们需要注意几个事项:

  1. 我们需要存储构造器方法传入的 $data 数组,以便后续我们可以从中获取它的元素。

  2. 还需要一个内部索引(或指针)来跟踪 currentnext 元素。

  3. rewind() 仅仅重置 index 属性,这样 current()next() 才能正常工作。

  4. 键名并非只能是数字类型!这里使用数组索引是为了保证示例足够简单。

我们可以向下面这样运行这段代码:

$iterator = new MyIterator(["foo", "bar", "baz"]);
 
foreach ($iterator as $key =&gt; $value) {
    print "item: " . $key . "|" . $value . "\n";
}
登录后复制

这看起来需要处理太多工作,但是这是能够像数组一样使用 foreach/for 功能的一个简洁实现。

IteratorAggregate(聚合迭代器)

还记得第二个接口抛出的 Traversable 异常么?下面看一个比实现 Iterator 接口更快的实现吧:

class MyIteratorAggregate implements IteratorAggregate
{
    protected $data;

    public function __construct($data)
    {
        $this-&gt;data = $data;
    }

    public function getIterator()
    {
        return new ArrayIterator($this-&gt;data);
    }
}
登录后复制

这里我们作弊了。相比于实现一个完整的 Iterator,我们通过 ArrayIterator() 装饰。不过,这相比于通过实现完整的 Iterator 简化了不少代码。

4232843658-5b4d4c9d111eb_articlex[1].jpg

兄弟莫急!先让我们比较一些代码。首先,我们在不使用生成器的情况下从文件中读取每一行数据:

$content = file_get_contents(__FILE__);

$lines = explode("\n", $content);

foreach ($lines as $i =&gt; $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 =&gt; $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] =&gt; rewind
//     [1] =&gt; valid
//     [2] =&gt; current
//     [3] =&gt; key
//     [4] =&gt; next
//     [5] =&gt; send
//     [6] =&gt; throw
//     [7] =&gt; __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-&gt;current() . "\n";

$generator-&gt;send("bar");
登录后复制

数据通过 yield 关键字传入和返回。首先,执行 current() 代码直到遇到 yield,返回 foosend() 将输出传入到生成器打印输入的位置。你需要习惯这种用法。

抛出异常(Throw)

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

看看下面这个示例:

$multiply = function($x, $y) {
    yield $x * $y;
};

print $multiply(5, 6)-&gt;current(); // 30
登录后复制

现在让我们将它封装到另一个函数中:

$calculate = function ($op, $x, $y) use ($multiply) {
    if ($op === 'multiply') {
        $generator = $multiply($x, $y);

        return $generator-&gt;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-&gt;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-&gt;throw(new InvalidArgumentException());
        }

        return $generator-&gt;current();
    }
};
print $calculate('multiply', 5, 'foo'); // PHP Fatal error...
登录后复制

棒呆了!我们不仅可以像迭代器一样使用生成器。还可以通过它们发送数据并抛出异常。它们是可中断和可恢复的函数。有些语言把这些函数叫做……

405043990-5b4d4c9d0d35a_articlex[1].jpg

我们可以使用协程(coroutines)来构建异步代码。让我们来创建一个简单的任务调度程序。首先我们需要一个 Task 类:

class Task
{
    protected $generator;

    public function __construct(Generator $generator)
    {
        $this-&gt;generator = $generator;
    }

    public function run()
    {
        $this-&gt;generator-&gt;next();
    }

    public function finished()
    {
        return !$this-&gt;generator-&gt;valid();
    }
}
登录后复制

Task 是普通生成器的装饰器。我们将生成器赋值给它的成员变量以供后续使用,然后实现一个简单的 run()finished() 方法。run() 方法用于执行任务,finished() 方法用于让调度程序知道何时终止运行。

然后我们需要一个 Scheduler 类:

class Scheduler
{
    protected $queue;

    public function __construct()
    {
        $this-&gt;queue = new SplQueue();
    }

    public function enqueue(Task $task)
    {
        $this-&gt;queue-&gt;enqueue($task);
    }

    pulic function run()
    {
        while (!$this-&gt;queue-&gt;isEmpty()) {
            $task = $this-&gt;queue-&gt;dequeue();
            $task-&gt;run();

            if (!$task-&gt;finished()) {
                $this-&gt;queue-&gt;enqueue($task);
            }
        }
    }
}
登录后复制

Scheduler 用于维护一个待执行的任务队列。run() 会弹出队列中的所有任务并执行它,直到运行完整个队列任务。如果某个任务没有执行完毕,当这个任务本次运行完成后,我们将再次入列。

SplQueue 对于这个示例来讲再合适不过了。它是一种 FIFO(先进先出:fist in first out) 数据结构,能够确保每个任务都能够获取足够的处理时间。

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

$scheduler = new Scheduler();

$task1 = new Task(call_user_func(function() {
    for ($i = 0; $i enqueue($task1);
$scheduler-&gt;enqueue($task2);

$scheduler-&gt;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-&gt;generator = $generator;
    }

    public function run()
    {
        if ($this-&gt;run) {
            $this-&gt;generator-&gt;next();
        } else {
            $this-&gt;generator-&gt;current();
        }

        $this-&gt;run = true;
    }

    public function finished()
    {
        return !$this-&gt;generator-&gt;valid();
    }
}
登录后复制

我们需要调整首次 run() 方法调用,从生成器当前有效的指针读取运行。后续调用可以从下一个指针读取运行...

3991232520-5b4d4c9d0f4b0_articlex[1].jpg

有些人基于这个思路实现了一些超赞的类库。我们来看看其中的两个...

RecoilPHP

RecoilPHP 是一套基于协程的类库,它最令人印象深刻的是用于 ReactPHP 内核。可以将事件循环在 RecoilPHP 和 RecoilPHP 之间进行交换,而你的程序无需架构上的调整。

我们来看一下 ReactPHP 异步 DNS 解决方案:

function resolve($domain, $resolver) {
    $resolver
        -&gt;resolve($domain)
        -&gt;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-&gt;create("8.8.8.8", $loop);
 
    resolve("silverstripe.org", $resolver);
    resolve("wordpress.org", $resolver);
    resolve("wardrobecms.com", $resolver);
    resolve("pagekit.com", $resolver);
 
    $loop-&gt;run();
}
 
run();
登录后复制

resolve() 接收域名和 DNS 解析器,并使用 ReactPHP 执行标准的 DNS 查找。不用太过纠结与 resolve() 函数内部。重要的是这个函数不是生成器,而是一个函数!

run() 创建一个 ReactPHP 事件循环,DNS 解析器(这里是个工厂实例)解析若干域名。同样,这个也不是一个生成器。

想知道 RecoilPHP 到底有何不同?还希望掌握更多细节!

use Recoil\Recoil;
 
function resolve($domain, $resolver)
{
    try {
        $ip = (yield $resolver-&gt;resolve($domain));
 
        print "domain: " . $domain . "\n";
        print "ip: " . $ip . "\n";
    } catch (Exception $exception) {
        print $exception-&gt;getMessage() . "\n";
    }
}
 
function run()
{
    $loop = (yield Recoil::eventLoop());
 
    $factory = new React\Dns\Resolver\Factory();
 
    $resolver = $factory-&gt;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 (&amp;$clients) {
            $clients-&gt;attach($client);
             
            $host = $client-&gt;getRemoteAddress();
            $port = $client-&gt;getRemotePort();
             
            $name = $host . ":" . $port;
             
            try {
                foreach ($clients as $stream) {
                    if ($client !== $stream) {
                        $stream-&gt;write($name . "connected.\n");
                    }
                }
 
                yield $client-&gt;write("Welcome " . $name . "!\n");
                 
                while ($client-&gt;isReadable()) {
                    $data = trim(yield $client-&gt;read());
                     
                    if ("/exit" === $data) {
                        yield $client-&gt;end("Goodbye!\n");
                    } else {
                        $message = $name . ":" . $data . "\n";
                        
                        foreach ($clients as $stream) {
                            if ($client !== $stream) {
                                $stream-&gt;write($message);
                            }
                        }
                    }
                }
            } catch (Exception $exception) {
                $client-&gt;close($exception);
            } finally {
                $clients-&gt;detach($client);
                foreach ($clients as $stream) {
                    $stream-&gt;write($name . "disconnected.\n");
                }
            }
        }
    );
     
    while ($server-&gt;isOpen()) {
        $handler(yield $server-&gt;accept());
    }
}, $factory-&gt;create("127.0.0.1", 6000));
 
Loop::run();
登录后复制

据我所知,这段代码所做的事情如下:

  1. 在 127.0.0.1 和 6000 端口创建一个服务器实例,然后将其传入外部生成器.

  2. 外部生成器运行,同时服务器等待新连接。当服务器接收一个连接它将其传入内部生成器。

  3. 内部生成器写入消息到 socket。当 socket 可读时运行。

  4. 每次 socket 向服务器发送消息时,内部生成器检测消息是否是退出标识。如果是,通知其他 socket。否则,其它 socket 发送这个相同的消息。

打开命令行终端输入 nc localhost 6000 查看执行结果!

该示例使用 SplObjectStorage 跟踪 socket 连接。这样我们就可以向所有 socket 发送消息。

4267668675-5b4d4c9d11821_articlex[1].jpg

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

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

相关推荐:

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

以上就是PHP的多任务协程处理的分析的详细内容,更多请关注php中文网其它相关文章!

智能AI问答
PHP中文网智能助手能迅速回答你的编程问题,提供实时的代码和解决方案,帮助你解决各种难题。不仅如此,它还能提供编程资源和学习指导,帮助你快速提升编程技能。无论你是初学者还是专业人士,AI智能助手都能成为你的可靠助手,助力你在编程领域取得更大的成就。
来源:php中文网
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系admin@php.cn
最新问题
热门教程
更多>
最新下载
更多>
网站特效
网站源码
网站素材
前端模板
关于我们 免责申明 意见反馈 讲师合作 广告合作 最新更新
php中文网:公益在线php培训,帮助PHP学习者快速成长!
关注服务号 技术交流群
PHP中文网订阅号
每天精选资源文章推送
PHP中文网APP
随时随地碎片化学习
PHP中文网抖音号
发现有趣的

Copyright 2014-2024 https://www.php.cn/ All Rights Reserved | php.cn | 湘ICP备2023035733号