基于swoole的swoolefy實(shí)現(xiàn)類似go的waitGroup多并發(fā)協(xié)程調(diào)度

ZweiZhao 發(fā)布于2019-07-01 12:44 / 1407人閱讀

摘要：基于實(shí)用，抽象事件處理類，實(shí)現(xiàn)與底層的回調(diào)的解耦，支持協(xié)程調(diào)度，同步異步調(diào)用，全局事件注冊(cè)，心跳檢查，異步任務(wù)，多進(jìn)程池等，內(nèi)置等常用組件等。

swoolefy是一個(gè)基于swoole實(shí)現(xiàn)的輕量級(jí)高性能的常駐內(nèi)存型的API和Web應(yīng)用服務(wù)框架，高度封裝了http，websocket，udp服務(wù)器，以及基于tcp實(shí)現(xiàn)可擴(kuò)展的rpc服務(wù)，同時(shí)支持composer包方式安裝部署項(xiàng)目。基于實(shí)用，swoolefy抽象Event事件處理類，實(shí)現(xiàn)與底層的回調(diào)的解耦，支持協(xié)程調(diào)度，同步|異步調(diào)用，全局事件注冊(cè)，心跳檢查，異步任務(wù)，多進(jìn)程(池)等，內(nèi)置view、log、session、mysql、redis、mongodb等常用組件等。

目前swoolefy4.2+版本完全支持swoole4.2.13+的協(xié)程，推薦使用swoole4.3+

GitHub：https://github.com/bingcool/s...

下面主要講解一下如何實(shí)現(xiàn)了類似go的waitGroup的功能
1、定義GoWaitGroup的類：


+----------------------------------------------------------------------
 */

namespace SwoolefyCore;

use SwooleCoroutineChannel;

class GoWaitGroup {
    /**
     * @var int
     */
    private $count = 0;

    /**
     * @var Channel
     */
    private $chan;

    /**
     * @var array
     */
    private $result = [];

    /**
     * WaitGroup constructor
     */
    public function __construct() {
        $this->chan = new Channel;
    }

    /**
     * add
     */
    public function go(Closure $go_func = null) {
        $this->count++;
        if($go_func instanceof Closure) {
            go($go_func);
        }
    }

    /**
     * start
     */
    public function start() {
        $this->count++;
        return $this->count;
    }

    /**
     * done
     */
    public function done(string $key, $data = null) {
        if(!empty($data)) {
            $this->result[$key] = $data;
        }
        $this->chan->push(1);
    }

    /**
     * wait
     */
    public function wait() {
        while($this->count--) {
            $this->chan->pop();
        }
        $result = $this->result;
        $this->result = [];
        $this->count = 0;
        return $result;
    }

}

2、在swoolefy中調(diào)用

class GroupController extends BController {
    public function waitgroup() {
        // 創(chuàng)建一個(gè)waitGroup實(shí)例
        $wg = new SwoolefyCoreGoWaitGroup();
         
         // 第一種方式，直接$wg->go()函數(shù)中執(zhí)行g(shù)o的協(xié)程函數(shù)
        $wg->go(function() use ($wg) {
            // 掛起協(xié)程
            $fp = stream_socket_client("tcp://www.baidu.com:80", $errno, $errstr, 30);
            // 協(xié)程返回的數(shù)據(jù)
            $wg->done("mysql", "mysql");
        });

        $wg->go(function() use ($wg) {
            sleep(1);
            $wg->done("tengxun", "weixin and qq");
        });

        //掛起當(dāng)前協(xié)程，等待所有任務(wù)完成后恢復(fù)
        //$result = $wg->wait();
        //這里 $result 包含了 1 個(gè)任務(wù)執(zhí)行結(jié)果
        //var_dump($result);
        
        //第二種方式，添加$wg->start()，啟動(dòng)協(xié)程，然后使用swoole的原生go執(zhí)行協(xié)程函數(shù)
        $wg->start();
        go(function () use ($wg) {
            // 掛起協(xié)程
            sleep(1);
            $wg->done("taobao", "ali baba");
        });
        
         //第二種方式，添加$wg->start()，啟動(dòng)協(xié)程，然后使用swoole的原生go執(zhí)行協(xié)程函數(shù)
        $wg->start();
        go(function () use ($wg) {
            // 掛起協(xié)程
            sleep(1);
            $wg->done("baidu", "baidu");
        });
        // 以上三個(gè)協(xié)程將會(huì)并發(fā)調(diào)用，wait()函數(shù)實(shí)現(xiàn)等待三個(gè)協(xié)程數(shù)據(jù)返回
        //掛起當(dāng)前協(xié)程，讓出cpu控制權(quán)，cpu可以做其他的事情，直到待所有任務(wù)完成后恢復(fù)
        $result = $wg->wait();
        //這里 $result 包含了 2 個(gè)任務(wù)執(zhí)行結(jié)果
        var_dump($result);
    }
}

至此一個(gè)最簡(jiǎn)單的并發(fā)調(diào)用就完成了，你可以愉快使用gowaitGroup的協(xié)程調(diào)用了

GPU云服務(wù)器云服務(wù)器基于webrtc的電話回聲消除的研究與實(shí)現(xiàn) 基于webrtc的視頻會(huì)議系統(tǒng)的設(shè)計(jì)與實(shí)現(xiàn) 多進(jìn)程實(shí)現(xiàn)并發(fā)服務(wù)器多線程實(shí)現(xiàn)并發(fā)服務(wù)器