中文在线一区二区_欧美在线综合_久久久久久综合_欧美一区二区三区视频_国产免费看_国产福利精品一区

如何實(shí)現(xiàn)隊(duì)列消費(fèi)/自定義進(jìn)程

可能我們會(huì)經(jīng)常遇見需要不斷消費(fèi)隊(duì)列內(nèi)內(nèi)容的場(chǎng)景，我們以EasySwoole中自定義進(jìn)程的方式，來(lái)實(shí)現(xiàn)這一功能。

實(shí)現(xiàn)代碼

定義消費(fèi)進(jìn)程邏輯

<?php
/**
 * Created by PhpStorm.
 * User: Tioncico
 * Date: 2018/10/18 0018
 * Time: 9:43
 */

namespace App\Process;

use EasySwoole\Component\Process\AbstractProcess;
use Swoole\Process;

class Consumer extends AbstractProcess
{
    private $isRun = false;
    public function run($arg)
    {
        // TODO: Implement run() method.
        /*
         * 舉例，消費(fèi)redis中的隊(duì)列數(shù)據(jù)
         * 定時(shí)500ms檢測(cè)有沒有任務(wù)，有的話就while死循環(huán)執(zhí)行
         */
        $this->addTick(500,function (){
            if(!$this->isRun){
                $this->isRun = true;
                $redis = new \redis();//此處為偽代碼，請(qǐng)自己建立連接或者維護(hù)redis連接
                while (true){
                    try{
                        $task = $redis->lPop('task_list');
                        if($task){
                            // do you task
                        }else{
                            break;
                        }
                    }catch (\Throwable $throwable){
                        break;
                    }
                }
                $this->isRun = false;
            }
            var_dump($this->getProcessName().' task run check');
        });
    }

    public function onShutDown()
    {
        // TODO: Implement onShutDown() method.
    }

    public function onReceive(string $str, ...$args)
    {
        // TODO: Implement onReceive() method.
    }
}

注冊(cè)消費(fèi)進(jìn)程

在EasySwoole的全局事件中，注冊(cè)消費(fèi)進(jìn)程。

<?php
use App\Process\TestProcess;
use EasySwoole\Component\Process\Manager;
use EasySwoole\EasySwoole\Swoole\EventRegister;

public static function mainServerCreate(EventRegister $register)
{

    $allNum = 3;
    for ($i = 0 ;$i < $allNum;$i++){
        $processConfig= new \EasySwoole\Component\Process\Config();
        $processConfig->setProcessName('testProcess'.$i);//設(shè)置進(jìn)程名稱
        Manager::getInstance()->addProcess(new TestProcess($processConfig));
    }
}