Swoole 协程

[twn39]

多个协程并行，类似于nodejs的Promise.all()。

<?php
/**
 * Created by PhpStorm.
 * User: weinan
 * Date: 2019/2/23
 * Time: 21:36
 *
 * 首先确保所执行的函数体里面没有阻塞的函数，例如PDO，file_get_contents等等，
 * 如果有这些函数，请先执行： Swoole\Runtime::enableCoroutine();
 */

use Swoole\Coroutine as Co;

class GroupWait
{
    private $channel;

    private $coList;

    private $len;

    public function __construct(array $coList)
    {
        $this->coList = $coList;
        $this->len = count($coList);
        $this->channel = new Co\channel($this->len);
    }

    public function wait()
    {
        foreach ($this->coList as $key => $co) {
            go(function () use ($key, $co) {
                $data = $co();
                $this->channel->push([$key, $data]);
            });
        }

        $result = [];
        for($i = 1; $i <= $this->len; $i++) {
            list($k, $v) = $this->channel->pop();
            $result[$k] = $v;
        }

        return $result;
    }

    public function __destruct()
    {
        $this->channel->close();
    }

}

[twn39]

使用方式：

go(function () {

    $fetchWeb = function () {
        return  file_get_contents('http://www.baidu.com/');
    };

    $list = [
        $fetchWeb,
        $fetchWeb,
        $fetchWeb,
        $fetchWeb,
        $fetchWeb,
        $fetchWeb,
        $fetchWeb,
        $fetchWeb,
        $fetchWeb,
        $fetchWeb,
        $fetchWeb,
        $fetchWeb,
        $fetchWeb,
        $fetchWeb,
        $fetchWeb,
        $fetchWeb,
        $fetchWeb,
        $fetchWeb,
        $fetchWeb,
        $fetchWeb,
        $fetchWeb,
        $fetchWeb,
        $fetchWeb,
        $fetchWeb,
    ];

    $gw = new GroupWait($list);

    $result = $gw->wait();
    var_dump($result);
});

这里用了file_get_contents,所以需要在头部加上： Swoole\Runtime::enableCoroutine();

[twn39]

<?php

use Swoole\Coroutine as Co;

class Group
{

    private $channel;

    private $items = [];

    private $count;

    public function __construct()
    {
        $this->channel = new Swoole\Coroutine\Channel();
    }

    public function add(callable $cor)
    {
        $this->items[] = $cor;
    }

    public function run()
    {
        $this->count = count($this->items);
        foreach ($this->items as $item) {
            go(function () use ($item) {
                $item();
                $this->channel->push(true);
            });
        }
        while ($this->count --) {
            $this->channel->pop();
        }
    }
}

go(function () {
    $g = new Group();

    $g->add(function () {
        Co::sleep(5);
        echo time();
        echo "sleep: 5\r\n";
    });
    $g->add(function () {
        Co::sleep(3);
        echo time();
        echo "sleep: 3\r\n";
    });
    $g->add(function () {
        Co::sleep(8);
        echo time();
        echo "sleep: 8\r\n";
    });

    $g->run();

});