apisongcan/extend/queue.php

<?php
declare(strict_types=0);

/**
 * 队列处理
 *
 *
 * @package    
 */

class QueueClient
{
    private static $queuedb;

    public static function push($queue_name,$host,$port,$key, $value)
    {
        if (!is_object(self::$queuedb)) {
            self::$queuedb = new QueueDB($queue_name,$host,$port);
        }
        return self::$queuedb->push(serialize([$key=>$value]));
    }
}

class QueueServer
{
    private $_queuedb;
    
    public function __construct($queue_name,$host,$port) {
        $this->_queuedb = new QueueDB($queue_name,$host,$port);
    }

    public function pop($key,$time)
    {
        $result = $this->_queuedb->pop($key,$time);
        if($result != null) {
            return unserialize($result);
        } else {
            return false;
        }
    }

    public function scan() {
        return $this->_queuedb->scan();
    }
}

class QueueDB
{
    private $_redis;

    private $queue_name = '';
    //存定义存储表的数量,系统会随机分配存储

    public function __construct($name,$host,$port)
    {
        $this->queue_name = $name;
        if ( !extension_loaded('redis') ) {
            return new Exception('redis failed to load');
        }
        $this->_redis = new Redis();
        $this->_redis->connect($host,$port,20);
        $this->_redis->setOption(Redis::OPT_READ_TIMEOUT, 10);
    }

    public function push($value)
    {
        try {
            return $this->_redis->lPush($this->queue_name,$value);
        } catch(Exception $e) {
             return new Exception($e->getMessage());
        }
    }

    public function scan()
    {
        return $this->queue_name;
    }

    public function pop($key, $time)
    {
        $result = $this->_redis->brPop($key, $time);
        if ($result) {
            return $result[1];
        } else {
            return null;
        }
    }

    public function clear() {
        $this->_redis->flushAll();
    }
}