五月天激情丁香,国产精品2019,国产成人精品亚洲2020,国产精品免费视频一区二区三区,开心久久婷婷综合中文字幕,天堂视频在线观看免费完整版

Queue 介紹

原理

EasySwoole 封裝實現了一個輕量級的隊列，默認以 Redis 作為隊列驅動器。

可以自己實現一個隊列驅動來實現隊列，用 kafka 作為隊列驅動器或者 其他驅動器方式 作為隊列驅動器，來進行存儲。

從上可知，Queue 并不是一個單獨使用的組件，它更像一個對不同驅動的隊列進行統一封裝的門面組件。

組件要求

  • ext-swoole: >=4.4.0
  • easyswoole/component: ^2.0
  • easyswoole/redis-pool: ~2.2.0

安裝方法

composer require easyswoole/queue=2.1.x

倉庫地址

easyswoole/queue

基本使用

  • 注冊隊列驅動器
  • 設置消費進程
  • 生產者投遞任務

定義一個隊列

<?php

namespace App\Utility;

use EasySwoole\Component\Singleton;
use EasySwoole\Queue\Queue;

class MyQueue extends Queue
{
    use Singleton;
}

定義消費進程

<?php

namespace App\Utility;

use EasySwoole\Component\Process\AbstractProcess;
use EasySwoole\Queue\Job;

class QueueProcess extends AbstractProcess
{
    protected function run($arg)
    {
        go(function () {
            MyQueue::getInstance()->consumer()->listen(function (Job $job) {
                // 打印消費數據
                var_dump($job->getJobData());
            });
        });
    }
}

可以多進程，多協程消費

驅動注冊

<?php

namespace EasySwoole\EasySwoole;

use EasySwoole\Component\Timer;
use EasySwoole\EasySwoole\Swoole\EventRegister;
use EasySwoole\EasySwoole\AbstractInterface\Event;
use EasySwoole\Http\Request;
use EasySwoole\Http\Response;
use EasySwoole\Queue\Job;

class EasySwooleEvent implements Event
{

    public static function initialize()
    {
        // TODO: Implement initialize() method.
        date_default_timezone_set('Asia/Shanghai');
    }

    public static function mainServerCreate(EventRegister $register)
    {
        // redis-pool 的使用請看 redis 章節文檔(http://www.fe88.cn/Components/Redis/pool.html)
        // 注冊一個名為 queue 的 Redis 連接池
        \EasySwoole\RedisPool\RedisPool::getInstance()->register(new \EasySwoole\Redis\Config\RedisConfig(
            [
                'host' => '127.0.0.1',
                'port' => '6379',
                // [可選參數] 密碼
                // 'auth' => ''
            ]
        ), 'queue');

        // 獲取 Redis 連接池中的一個 Redis 連接對象
        $redisPool = \EasySwoole\RedisPool\RedisPool::getInstance()->getPool('queue');

        // 配置隊列驅動器，底層使用 Redis 驅動，并設置隊列名為 'queue'
        $driver = new \EasySwoole\Queue\Driver\Redis($redisPool, 'queue');

        // 注冊自定義隊列
        \App\Utility\MyQueue::getInstance($driver);

        // 注冊一個消費進程
        \EasySwoole\Component\Process\Manager::getInstance()->addProcess(new \App\Utility\QueueProcess());

        // 模擬生產者，投遞任務到隊列中，可以在任意位置投遞
        $register->add($register::onWorkerStart, function ($server, $id) {
            if ($id == 0) {
                Timer::getInstance()->loop(3000, function () {
                    $job = new Job();
                    // 設置投遞的隊列任務數據
                    $job->setJobData(['time' => \time()]);
                    \App\Utility\MyQueue::getInstance()->producer()->push($job);
                });
            }
        });
    }
}

關于進程安全退出問題請看 進程章節。

進階使用

我們可以自定義驅動，實現 RabbitMQ、Kafka 等消費隊列軟件的封裝。

用戶需要定義類，并實現 \EasySwoole\Queue\QueueDriverInterface 接口的幾個方法即可。該接口的詳細實現請看下文。

QueueDriverInterface 接口類實現

<?php

namespace EasySwoole\Queue;

interface QueueDriverInterface
{
    public function push(Job $job):bool ;

    public function pop(float $timeout = 3.0):?Job;

    public function size():?int ;
}

組件自帶的 Redis 隊列驅動器實現

<?php

namespace EasySwoole\Queue\Driver;

use EasySwoole\Queue\Job;
use EasySwoole\Queue\QueueDriverInterface;
use EasySwoole\Redis\Redis as Connection;
use EasySwoole\RedisPool\Pool;

class Redis implements QueueDriverInterface
{

    protected $pool;
    protected $queueName;
    public function __construct(Pool $pool,string $queueName = 'easy_queue')
    {
        $this->pool = $pool;
        $this->queueName = $queueName;
    }

    public function push(Job $job): bool
    {
        $data = serialize($job);
        return $this->pool->invoke(function (Connection $connection)use($data){
            return $connection->lPush($this->queueName,$data);
        });
    }

    public function pop(float $timeout = 3.0): ?Job
    {
        return $this->pool->invoke(function (Connection $connection){
            $data =  $connection->rPop($this->queueName);
            if($data){
                return unserialize($data);
            }
            return null;
        });
    }

    public function size(): ?int
    {
        return $this->pool->invoke(function (Connection $connection){
            return $connection->lLen($this->queueName);
        });
    }
}

Queue 多節點使用

定義第一個隊列(自定義 nodeId)

<?php

namespace App\Utility;

use EasySwoole\Component\Singleton;
use EasySwoole\Queue\Queue;
use EasySwoole\Queue\QueueDriverInterface;

class MyQueue1 extends Queue
{
    use Singleton;

    public function __construct(QueueDriverInterface $driver)
    {
        parent::__construct($driver);
        // 自定義 nodeId
        $this->setNodeId('xxxxx1');
    }
}

定義第二個隊列(自動生成 nodeId)

<?php

namespace App\Utility;

use EasySwoole\Component\Singleton;
use EasySwoole\Queue\Queue;

class MyQueue2 extends Queue
{
    use Singleton;
}

獲取節點id

<?php

namespace App\Utility;

use EasySwoole\Component\Process\AbstractProcess;
use EasySwoole\Queue\Job;

class QueueProcess extends AbstractProcess
{
    protected function run($arg)
    {   
        // 消費隊列
        go(function () {
            MyQueue1::getInstance()->consumer()->listen(function (Job $job) {
                // 打印 節點Id
                var_dump($job->getNodeId());
                // 打印 任務Id
                var_dump($job->getJobId());
            });
            MyQueue2::getInstance()->consumer()->listen(function (Job $job) {
                // 打印 節點Id
                var_dump($job->getNodeId());
                // 打印 任務Id
                var_dump($job->getJobId());
            });
        });
    }
}

可以多進程，多協程消費

驅動注冊

<?php

namespace EasySwoole\EasySwoole;

use App\Utility\QueueProcess;
use EasySwoole\Component\Timer;
use EasySwoole\EasySwoole\Swoole\EventRegister;
use EasySwoole\EasySwoole\AbstractInterface\Event;
use EasySwoole\Http\Request;
use EasySwoole\Http\Response;
use EasySwoole\Queue\Job;

class EasySwooleEvent implements Event
{

    public static function initialize()
    {
        // TODO: Implement initialize() method.
        date_default_timezone_set('Asia/Shanghai');
    }

    public static function mainServerCreate(EventRegister $register)
    {
        // redis-pool 的使用請看 redis 章節文檔(http://www.fe88.cn/Components/Redis/pool.html)
        // 注冊一個名為 queue 的 Redis 連接池
        \EasySwoole\RedisPool\RedisPool::getInstance()->register(new \EasySwoole\Redis\Config\RedisConfig(
            [
                'host' => '127.0.0.1',
                'port' => '6379',
                // [可選參數] 密碼
                // 'auth' => ''
            ]
        ), 'queue');

        // 獲取 Redis 連接池中的一個 Redis 連接對象
        $redisPool = \EasySwoole\RedisPool\RedisPool::getInstance()->getPool('queue');

        // 配置隊列驅動器，底層使用 Redis 驅動，并設置隊列名為 'queue'
        $driver = new \EasySwoole\Queue\Driver\Redis($redisPool, 'queue');

        // 【這里是重點】
        // 注冊自定義隊列
        \App\Utility\MyQueue1::getInstance($driver);
        \App\Utility\MyQueue2::getInstance($driver);

        // 注冊一個消費進程
        \EasySwoole\Component\Process\Manager::getInstance()->addProcess(new \App\Utility\QueueProcess());

        // 模擬生產者，投遞任務到隊列中，可以在任意位置投遞
        $register->add($register::onWorkerStart, function ($server, $id) {
            if ($id == 0) {
                Timer::getInstance()->loop(3000, function () {
                    $job = new Job();
                    // 設置投遞的隊列任務數據
                    $job->setJobData(['time' => \time()]);
                    // 這里是重點
                    \App\Utility\MyQueue1::getInstance()->producer()->push($job);
                    \App\Utility\MyQueue2::getInstance()->producer()->push($job);
                });
            }
        });
    }
}

相關倉庫

EasySwoole 中利用 Redis 實現消息隊列

如何利用 EasySwoole 多進程多協程 Redis 隊列實現爬蟲

主站蜘蛛池模板： 五月激情婷婷网 | www.亚洲成在线 | 六月婷婷色 | 久草天堂 | 国产精品综合一区二区三区 | 人成电影网在线观看免费 | 亚洲青青青网伊人精品 | 成人久久免费视频 | 国产综合免费视频 | 99re热视频精品首页 | 久久99国产精品视频 | 精品欧美一区二区三区精品久久 | 天天草综合网 | 日本精品久久久久久久久免费 | 国产精品久久久久久久免费 | 激情五月激情综合网 | 国产视频自拍一区 | 老色99久久九九精品尤物 | 国产干美女 | 九七电影网97理论片 | 99久久久精品 | 欧美精品三区 | 色在线免费视频 | 欧美在线看欧美高清视频免费 | www.奇米影视.com | 四虎影视库永久地址发布 | 国产精品久久久久久一区二区三区 | 精品无人区乱码1区2区 | 青青草91久久国产频道 | 99热这里只有精品一区二区三区 | 精品视频一区二区三区四区 | 国产69精品久久久久99不卡 | 色中色在线视频 | 国产日韩欧美中文字幕 | 激情六月天婷婷 | 四虎精品影视 | 人人综合网 | 伊人网综合视频 | 成人黄色在线 | 五月激情婷婷综合 | 亚洲黄色激情网 |