协程（Coroutine）（二）

协程（Coroutine）（二）

1、并发查询

     关于协程的使用，经常会用到它来做并发查询，可以用协程+waitgroup+channel的方式来做，也可以用csp并发来做。

     1）协程+waitgroup+channel

go(function (){
       $chan=new Channel(12);//设置通道的容量是12
       $wait=new WaitGroup();
       for ($i=1;$i<=12;$i++){
            $wait->add();
            go(function ()use($wait,$chan,$i){
                 co::sleep(rand(1,3));
                  $chan->push("第{$i}个月的数据！");
                  $wait->done();
             });
       }
       $wait->wait();//挂起当前协程，等待所有任务完成后执行下面代码
       while(true){
            if ($chan->isEmpty()){
                 break;
             }
             $res=$chan->pop();
             Logger::getInstance()->info($res.PHP_EOL);
       }
       Logger::getInstance()->info("------------------".PHP_EOL);
 });

    运行结果：

      2）csp并发

public function testCsp1()
{
       $time=microtime(true);
       $chan=new Channel();
       go(function ()use($chan){
            $csp = new EasySwooleComponentCsp();
            $csp->add('t1',function (){
                co::sleep(4);
                return 't1 result';
            });
            $csp->add('t2',function (){
                co::sleep(3);
                return 't2 result';
            });
            $res=$csp->exec();
            $chan->push($res);
        });
        $time1=microtime(true)-$time;
        Logger::getInstance()->info("go代码执行后用时：".$time1."s");
        $res=$chan->pop();
        $time2=microtime(true)-$time;
        Logger::getInstance()->info("获取到结果用时：".$time2."s");
        Logger::getInstance()->info('res:'.json_encode($res));
}

       运行结果：

2、csp使用go和不使用go的区别

    上面2）中的csp并发是是在协程容器下执行的。作为比较，下面的代码展示在非协程容器下来执行

public function testCsp2()
{
        $time1=microtime(true);
        $csp = new EasySwooleComponentCsp();
        $csp->add('t1',function (){
            co::sleep(4);
            return 't1 result';
        });
        $csp->add('t2',function (){
            co::sleep(3);
             return 't2 result';
        });
        $res=$csp->exec();
        $time2=microtime(true);
        $time=$time2-$time1;
        Logger::getInstance()->info("用时：".$time."s");
        Logger::getInstance()->info("res:".json_encode($res));
}

    运行结果：

    总结

    1）可以看出在go()里面执行的代码，是不会阻塞下面的代码运行的，最后$csp->exec()获取到结果才返回。没有在go()里面写的代码，就是正常从上往下执行，执行到$csp->exec()这段代码时，因为是相对比较耗时的任务，要获取到结果才会接着往下执行。

    2）协程之间的通信需要使用channel方式，协程中访问外部变量必须使用use关键字,或者传形参方式,不能引用变量。外部要访问协程中的变量需要通过channel的方式。

    3）注意：exec方法提供了一个默认参数：超时时间(默认为5s)，当调用$csp->exec()后，最长等待5s左右会返回结果。如果你在t2函数中co::sleep(6)，那么5s后，返回的数据中不会包含t2函数的返回数据。

3、csp类的源码不多，我们来看看

     个人的理解都放到注释里面了，有问题的地方欢迎大家指正。

<?php

namespace EasySwooleComponent;
use SwooleCoroutineChannel;
use SwooleCoroutine;
class Csp
{
    private $chan;       //通道channel的实例
    private $count = 0;  //任务总数
    private $success = 0;//任务成功执行的总数
    private $task = [];  //任务数组

    function __construct(int $size = 8)
    {
        $this->chan = new Channel($size);
    }

    function add($itemName,callable $call):Csp
    {
        $this->count = 0;
        $this->success = 0;
       //将所有的请求的别名与回调函数以数组的形式存放在私有变量$task中
        $this->task[$itemName] = $call;
        return $this;
    }

    function successNum():int
    {
        return $this->success;
    }
    //最重要的执行方法
    function exec(?float $timeout = 5)
    {
        if($timeout <= 0){
            $timeout = PHP_INT_MAX;
        }
        $this->count = count($this->task);//总的任务数
        //将所有的请求的别名与回调函数进行遍历
        foreach ($this->task as $key => $call){
            //在协程容器中，通过执行回调函数获取结果，调用channel进行通信
            Coroutine::create(function ()use($key,$call){
                $data = call_user_func($call);
                $this->chan->push([
                    'key'=>$key,
                    'result'=>$data
                ]);
            });
        }
        $result = [];
        $start = microtime(true);
        while($this->count > 0)
        {
            $temp = $this->chan->pop(1);
            if(is_array($temp)){
                $key = $temp['key'];
                $result[$key] = $temp['result'];
                $this->count--;   //任务总数减一
                $this->success++; //任务成功执行的数量加1
            }
            //考虑超时的情况，如果某个任务超时，那么其返回的数据就不会被纳入并发查询的最终结果中。
            if(microtime(true) - $start > $timeout){
                break;
            }
        }
        return $result;
    }
}

参考链接：

https://www.easyswoole.com/Cn/Swoole/Coroutine/introduction.html