EasySwoole中利用redis实现消息队列

<?php
namespace App\Utility;


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

class RedisQueue extends Queue
{
    use Singleton;
}

namespace App\Process;


use App\Utility\RedisQueue;
use EasySwoole\Component\Process\AbstractProcess;
use EasySwoole\Queue\Job;

class QueueProcess extends AbstractProcess
{

    protected function run($arg)
    {
        // TODO: Implement run() method.
        go(function (){
            RedisQueue::getInstance()->consumer()->listen(function (Job $job){
                var_dump($job->toArray());
            });
        });
    }
}

//在EasySwooleEvent.php 里面注册驱动
public static function mainServerCreate(EventRegister $register)
    {
        // TODO: Implement mainServerCreate() method.
        //注册redis驱动队列
        $redisData = Config::getInstance()->getConf('REDIS');
        $redisConfig = new RedisConfig($redisData);
        $redis = new RedisPool($redisConfig);
        $driver = new \EasySwoole\Queue\Driver\Redis($redis);
        RedisQueue::getInstance($driver);
        //注册消费进程
        ServerManager::getInstance()->addProcess(new QueueProcess());
        $register->add($register::onWorkerStart,function ($ser,$id){
            if($id == 0){
                Timer::getInstance()->loop(3000,function (){
				//自动投递任务
                    $job = new Job();
                    $job->setJobData(['time' => time()]);
                    RedisQueue::getInstance()->producer()->push($job);
                });
            }
        });
	}

    public function index()
    {
		$job = new Job();
		$job->setJobData(['test'=>'测试','time'=>time()]); 
		var_dump(RedisQueue::getInstance()->producer()->push($job));
	}