Thinkphp6 think-queue redis 封装队列任务类

- 日理万妓 2021-09-08 19:34 1217阅读 0赞

### 配置文件queue.php ###

<?php  
// +----------------------------------------------------------------------  
// | ThinkPHP \[ WE CAN DO IT JUST THINK IT \]  
// +----------------------------------------------------------------------  
// | Copyright (c) 2006-2016 http://thinkphp.cn All rights reserved.  
// +----------------------------------------------------------------------  
// | Licensed ( http://www.apache.org/licenses/LICENSE-2.0 )  
// +----------------------------------------------------------------------  
// | Author: yunwuxin <448901948@qq.com>  
// +----------------------------------------------------------------------  
use think\\facade\\Env;

return \[  
    'default'     => 'redis',  
    'connections' => \[  
        'sync'     => \[  
            'type' => 'sync',  
        \],  
        'database' => \[  
            'type' => 'database',  
            'queue'  => 'default',  
            'table'  => 'jobs',  
        \],  
        'redis'    => \[  
            'type'     => 'redis',  
            'queue'      => 'defaultredis',  
            'host'       => Env::get('redis.redis\_hostname', '127.0.0.1'),  
            'port'       => Env::get('redis.port', 6379),  
            'password'   => Env::get('redis.redis\_password', ''),  
            'select'     => Env::get('redis.select', 0),  
            'timeout'    => 0,  
            'persistent' => false,  
        \],  
    \],  
    'failed'      => \[  
        'type'  => 'none',  
        'table' => 'failed\_jobs',  
    \],  
\];

### 测试文件加入队列PublicController.php ###

<?php

namespace app\\api\\controller\\demo;

use crmeb\\utils\\Queue;  
use crmeb\\jobs\\CeshiJob;

/\*\*  
 \* 公共类  
 \* Class PublicController  
 \* @package app\\api\\controller  
 \*/  
class PublicController  
\{  
    //加入队列  
    //http://meishu.local/api/demo/demoqueue  
    public function demoqueue()\{  
        $res = Queue::instance()->do('ceshiqueue')->job(CeshiJob::class)->name('myqueue')->push();  
        if($res)\{  
            return app('json')->success('加入队列成功！');  
        \}else\{  
            return app('json')->fail('加入队列失败！');  
        \}  
    \}

### 队列工具类Queue.php ###

<?php

namespace crmeb\\utils;

use crmeb\\traits\\ErrorTrait;  
use think\\facade\\Queue as QueueThink;  
use think\\facade\\Log;

/\*\*  
 \* Class Queue  
 \* @package crmeb\\utils  
 \* @method $this do(string $do) 设置任务执行方法  
 \* @method $this job(string $job) 设置任务执行类名  
 \* @method $this errorCount(int $errorCount) 执行失败次数  
 \* @method $this data(...$data) 执行数据  
 \* @method $this secs(int $secs) 延迟执行秒数  
 \* @method $this log($log) 记录日志  
 \*/  
class Queue  
\{

use ErrorTrait;

/\*\*  
     \* 任务执行  
     \* @var string  
     \*/  
    protected $do = 'doJob';

/\*\*  
     \* 默认任务执行方法名  
     \* @var string  
     \*/  
    protected $defaultDo;

/\*\*  
     \* 任务类名  
     \* @var string  
     \*/  
    protected $job;

/\*\*  
     \* 错误次数  
     \* @var int  
     \*/  
    protected $errorCount = 3;

/\*\*  
     \* 数据  
     \* @var array|string  
     \*/  
    protected $data;

/\*\*  
     \* 任务名  
     \* @var null  
     \*/  
    protected $queueName = null;

/\*\*  
     \* 延迟执行秒数  
     \* @var int  
     \*/  
    protected $secs = 0;

/\*\*  
     \* 记录日志  
     \* @var string|callable|array  
     \*/  
    protected $log;

/\*\*  
     \* @var array  
     \*/  
    protected $rules = \['do', 'data', 'errorCount', 'job', 'secs', 'log'\];

/\*\*  
     \* @var static  
     \*/  
    protected static $instance;

/\*\*  
     \* Queue constructor.  
     \*/  
    protected function \_\_construct()  
    \{  
        $this->defaultDo = $this->do;  
    \}

/\*\*  
     \* @return static  
     \*/  
    public static function instance()  
    \{  
        if (is\_null(self::$instance)) \{  
            self::$instance = new static();  
        \}  
        return self::$instance;  
    \}

/\*\*  
     \* 放入消息队列  
     \* @param array|null $data  
     \* @return mixed  
     \*/  
    public function push(?array $data = null)  
    \{  
        if (!$this->job) \{  
            return $this->setError('需要执行的队列类必须存在');  
        \}  
        $res = QueueThink::\{$this->action()\}(...$this->getValues($data));  
        if(!$res)\{  
            $res = QueueThink::\{$this->action()\}(...$this->getValues($data));  
            if(!$res)\{  
                Log::error('加入队列失败，参数：'.json\_encode($this->getValues($data)));  
            \}  
        \}  
        $this->clean();  
        return $res;  
    \}

/\*\*  
     \* 队列名称  
     \*/  
    public function name($queueName = null)  
    \{  
        $this->queueName = $queueName;  
        return $this;  
    \}  
      
    /\*\*  
     \* 清除数据  
     \*/  
    public function clean()  
    \{  
        $this->secs = 0;  
        $this->data = \[\];  
        $this->log = null;  
        $this->queueName = null;  
        $this->errorCount = 3;  
        $this->do = $this->defaultDo;  
    \}

/\*\*  
     \* 获取任务方式  
     \* @return string  
     \*/  
    protected function action()  
    \{  
        return $this->secs ? 'later' : 'push';  
    \}

/\*\*  
     \* 获取参数  
     \* @param $data  
     \* @return array  
     \*/  
    protected function getValues($data)  
    \{  
        $jobData\['data'\] = $data ?: $this->data;  
        $jobData\['do'\] = $this->do;  
        $jobData\['errorCount'\] = $this->errorCount;  
        $jobData\['log'\] = $this->log;  
        $jobData\['queueName'\] = $this->log;  
        if ($this->do != $this->defaultDo) \{  
            $this->job .= '@' . $this->do;  
        \}  
        if ($this->secs) \{  
            return \[$this->secs, $this->job, $jobData, $this->queueName\];  
        \} else \{  
            return \[$this->job, $jobData, $this->queueName\];  
        \}  
    \}

/\*\*  
     \* @param $name  
     \* @param $arguments  
     \* @return $this  
     \*/  
    public function \_\_call($name, $arguments)  
    \{  
        if (in\_array($name, $this->rules)) \{  
            if ($name === 'data') \{  
                $this->\{$name\} = $arguments;  
            \} else \{  
                $this->\{$name\} = $arguments\[0\] ?? null;  
            \}  
            return $this;  
        \} else \{  
            throw new \\RuntimeException('Method does not exist' . \_\_CLASS\_\_ . '->' . $name . '()');  
        \}  
    \}  
\}

### 队列任务接口类JobInterface.php ###

<?php

namespace crmeb\\interfaces;

use think\\queue\\Job;

interface JobInterface  
\{  
    public function fire(Job $job, $data): void;  
\}

### 消息队列基类BaseJob.php ###

<?php

namespace crmeb\\basic;

use crmeb\\interfaces\\JobInterface;  
use think\\facade\\Log;  
use think\\queue\\Job;

/\*\*  
 \* 消息队列基类  
 \* Class BaseJob  
 \* @package crmeb\\basic  
 \*/  
class BaseJob implements JobInterface  
\{

/\*\*  
     \* @param $name  
     \* @param $arguments  
     \*/  
    public function \_\_call($name, $arguments)  
    \{  
        $this->fire(...$arguments);  
    \}

/\*\*  
     \* @param Job $job  
     \* @param $data  
     \*/  
    public function fire(Job $job, $data): void  
    \{  
        try \{  
            $action = $data\['do'\] ?? 'doJob';//任务名  
            $infoData = $data\['data'\] ?? \[\];//执行数据  
            $errorCount = $data\['errorCount'\] ?? 0;//最大错误次数  
            $log = $data\['log'\] ?? null;  
            if (method\_exists($this, $action)) \{  
                $this->\{$action\}(...$infoData);  
            \} else \{  
                $job->delete();  
            \}  
        \} catch (\\Throwable $e) \{  
            $job->delete();  
        \}  
    \}

/\*\*  
     \* 打印出成功提示  
     \* @param $log  
     \* @return bool  
     \*/  
    protected function info($log)  
    \{  
        try \{  
            if (is\_callable($log)) \{  
                print\_r($log() . "\\r\\n");  
            \} else if (is\_string($log) || is\_array($log)) \{  
                print\_r($log . "\\r\\n");  
            \}  
        \} catch (\\Throwable $e) \{  
            print\_r($e->getMessage());  
        \}  
    \}

/\*\*  
     \* 任务失败执行方法  
     \* @param $data  
     \* @param $e  
     \*/  
    public function failed($data, $e)  
    \{

\}  
\}

### 队列执行类CeshiJob.php ###

<?php  
namespace crmeb\\jobs;

use crmeb\\basic\\BaseJob;  
use think\\facade\\Db;  
use think\\queue\\Job;

class CeshiJob extends BaseJob  
\{  
    public function ceshiqueue(Job $job,$data)  
    \{  
        $errorCount = $data\['errorCount'\] ?? 0;//最大错误次数  
        try \{  
            $arr\['name'\] = mt\_rand(111111,999999);  
            $res = Db::name('demo')->insert($arr);  
            if($res)\{  
                //删除任务  
                $job->delete();  
            \}else\{  
                if ($job->attempts() >= $errorCount && $errorCount) \{  
                    //删除任务  
                    $job->delete();  
                \} else \{  
                    //从新放入队列  
                    $job->release();  
                \}  
            \}  
        \} catch (\\Exception $e) \{  
            //删除任务  
            $job->delete();  
        \}  
    \}  
\}

### 执行命令： ###

php think queue:work --queue myqueue

或者php think queue:work --queue orderJobQueue --daemon

或者php think queue:work --queue 执行默认队列