【转】swoole的channel之waitgroup实现

作者: 白云飞 分类: swoole,网络知识 发布时间: 2018-12-04 14:59 阅读:

上一篇 swoole4.0协程初探中通过并发请求淘宝和百度的首页的简单示例,大概了解了一下swoole的协程是如何工作的,示例中我们是在http请求返回后,分别做了相关的逻辑处理,就有人提到,能否有类似于golang的sync.WaitGroup,等所有的协程数据返回了再做统一处理?

答案是肯定的,必需能支持,接下来出场的就是swoole的另一个重要特性:channel,下面我们就来通过代码了解一下,如何通过channel实现WaitGroup功能

<?php
class waitgroup
{

    private $count = 0;
    private $chan;

    /**
     * waitgroup constructor.
     * @desc 初始化一个channel
     */
    public function __construct()
    {
        $this->chan = new chan;
    }

    /**
     * @desc 计数+1
     * @调用时机:在起一个协程前
     */
    public function add()
    {
        $this->count++;
    }

    /**
     * @desc 协程处理完成时调用
     */
    public function done()
    {

        $this->chan->push(true);
    }

    /**
     * @desc 堵塞的等待所有的协程处理完成
     */
    public function wait()
    {
        for ($i = 0; $i < $this->count; $i++) {
            //调用pop方法时,如果没有数据,此协程会挂起
            //当往chan中push数据后,协程会被恢复
            $this->chan->pop();
        }
    }

}

这么简单的一串代码就实现了WaitGroup功能了,那之前的http代码可以这样改造:

<?php
require_once "waitgroup.php";

use Swoole\Coroutine\Http\Client;

//此方法记录执行时间
function timediff($time)
{
    return microtime(true) - $time;
}

//创建http server
$http = new Swoole\Http\Server("0.0.0.0", 9501);
$http->set([
    //"daemonize" => true,
    "worker_num" => 1,
]);
$http->on('request', function ($request, $response) {

    //浏览器会自动发起这个请求,这也是很多人碰到的一个问题:
    //为什么我浏览器打开网站,收到了两个请求?
    if ($request->server['path_info'] == '/favicon.ico') {
        $response->end('');
        return;
    }

    $time = microtime(true);
    $response->header("content-type", "text/html; charset=UTF-8");
    //定义一个数组,用于存储结果,方便统一输出
    $result = [];
    $result[] = "1. 接受请求,此处被执行, 第" . __LINE__ . "行, 时间" . $time . "<br/>";


    $wg = new waitgroup();


    //加入wait计数
    $wg->add();
    //启动第一个协程
    go(function () use ($response, $wg, &$result) {
        $time = microtime(true);
        $result[] = "2. 进入第一个协程,发起http请求taobao, 第" . __LINE__ . "行, 时间:" . $time . "<br/>";

        //启动一个协程客户端client,请求淘宝首页
        $cli = new Client('www.taobao.com', 443, true);
        $cli->setHeaders([
            'Host' => "www.taobao.com",
            "User-Agent" => 'Chrome/49.0.2587.3',
            'Accept' => 'text/html,application/xhtml+xml,application/xml',
            'Accept-Encoding' => 'gzip',
        ]);
        $cli->set(['timeout' => 1]);
        //调用get方法,协程挂起,
        $cli->get('/index.php');
        //会等待i/o数据返回,执行wg的done方法,表示协程数据已返回
        $result[] = "7. get回taobao数据,唤起协程,此处被执行, 第" . __LINE__ . "行, 执行时间" . timediff($time) . "<br/>";
        $cli->close();

        //放在协程的最后执行
        $wg->done();
    });
    //上面get挂起协程后,后立马执行这一行
    $result[] = "3 cli->get时挂起协程了,此处被执行,不会被阻塞, 第" . __LINE__ . "行, 时间:" . microtime(true) . "<br/>";

    //加入wait计数
    $wg->add();
    //启动第二个协程
    go(function () use ($response, $wg, &$result) {
        $time = microtime(true);
        $result[] = "4. 进入第二个协程,发起http请求baidu, 第" . __LINE__ . "行, 时间:" . $time . "<br/>";
        //启动一个协程客户端client,请求百度首页
        $cli = new Client('www.baidu.com', 443, true);
        $cli->setHeaders([
            'Host' => "www.baidu.com",
            "User-Agent" => 'Chrome/49.0.2587.3',
            'Accept' => 'text/html,application/xhtml+xml,application/xml',
            'Accept-Encoding' => 'gzip',
        ]);
        $cli->set(['timeout' => 1]);
        //调用get方法,协程挂起,
        $cli->get('/index.php');
        //会等待i/o数据返回,执行wg的done方法,表示协程数据已返回
        $result[] = "6. get回baidu数据,唤起协程,此处被执行,正常这个先返回,因为ping百度更快,说明两个协程也是并发执行的, 第" . __LINE__ . "行, 执行时间" . timediff($time) . "<br/>";
        $cli->close();

        //放在协程的最后执行
        $wg->done();
    });
    //第二个协程get时挂起,执行到这一步
    $result[] = "5 cli->get时挂起协程了,此处被执行,不会被阻塞, 第" . __LINE__ . "行, 时间:" . microtime(true) . "<br/>";

    //堵塞中,直到所有的协程都执行调用done, 才会继续往下执行
    $wg->wait();
    $result[] = "总执行时间" . timediff($time) . ", 可看出约等于最长请求的时间而不是所有时间之和,协程间是真正并发执行的";
    $response->end(implode("<br/>", $result));
});
$http->start();

最终执行结果:

其实channel可以push任何数据,上面的示例中,我们用了一个result变量来存储结果,并且通过&$result的方式引用至协程内部了,如果我们把waitgroup代码稍改造一下,可以更优雅和方便一些:

<?php
class waitgroup
{

    private $count = 0;
    private $chan;

    /**
     * waitgroup constructor.
     * @desc 初始化一个channel
     */
    public function __construct()
    {
        $this->chan = new chan;
    }

    /**
     * @desc 计数+1
     * @调用时机:在起一个协程前
     */
    public function add()
    {
        $this->count++;
    }

    /**
     * @param $data
     * @desc 协程处理完成时调用,把数据存入channel
     */
    public function done($data)
    {

        $this->chan->push($data);
    }

    /**
     * @desc 堵塞的等待所有的协程处理完成并返回结果
     */
    public function wait()
    {
        $result = [];
        for ($i = 0; $i < $this->count; $i++) {
            //调用pop方法时,如果没有数据,此协程会挂起
            //当往chan中push数据后,协程会被恢复
            $result[] = $this->chan->pop();
        }

        return $result;
    }

}

我们给done方法加了一个$data参数,可以把处理结果push到channel中,那我们在wait成功后,可以取到结果

swoole的channel还是非常强大的,可以Push任何数据,  利用channel,让协程间的通讯变的非常简单可靠, 快快查看原文,了解更多swoole channel的使用姿势吧。

 

原文地址:https://mp.weixin.qq.com/s/XeZFNra1Uk0NKe2HqVFSFA

如果觉得我的文章对您有用,请随意打赏。您的支持将鼓励我继续创作!