Thinkphp6 think-queue redis 执行异步任务

古城微笑少年丶 2021-09-08 19:32 1653阅读 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'      => 'default',  
            '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 think\\facade\\Config;  
use app\\Request;  
use crmeb\\services\\sms\\Sms;  
use think\\facade\\Event;  
use Baiy\\ThinkAsync\\Facade\\Async;  
use think\\facade\\Queue;  
use crmeb\\jobs\\DemoJob;

/\*\*  
 \* 公共类  
 \* Class PublicController  
 \* @package app\\api\\controller  
 \*/  
class PublicController  
\{  
    //加入队列  
    //http://meishu.local/api/demo/demoqueue  
    public function demoqueue()\{  
        //当前任务将由哪个类来负责处理。  
        //当轮到该任务时，系统将生成一个该类的实例，并默认调用其 fire 方法  
        $jobHandlerClassName = 'crmeb\\jobs\\DemoJob';  
          
        //当前任务归属的队列名称，如果为新队列，会自动创建  
        //php think queue:work --queue orderJobQueue  
        //php think queue:work --queue orderJobQueue --daemon  
        $jobQueueName = "orderJobQueue";  
          
        //数组数据  
        $orderData = \[  
            'id' => uniqid(),  
            'time' => time(),  
        \];  
          
        //将该任务推送到消息队列，等待对应的消费者去执行  
        //这里只是负责将数据添加到相应的队列名称的队列里，消费者与生产者并无联系  
        $isPushed = Queue::push($jobHandlerClassName , $orderData, $jobQueueName);  
          
        if( $isPushed !== false )\{  
            echo date('Y-m-d H:i:s') . " 队列添加成功";  
        \}else\{  
            echo '队列添加失败';  
        \}  
    \}  
\}

### 队列执行文件DemoJob.php ###

<?php

namespace crmeb\\jobs;

use think\\facade\\Db;  
use think\\queue\\Job;

class DemoJob  
\{  
    public function fire(Job $job, $data)  
    \{  
        //有些消息在到达消费者时,可能已经不再需要执行了  
        $isJobStillNeedToBeDone = $this->checkDatabaseToSeeIfJobNeedToBeDone($data);  
        if(!$isJobStillNeedToBeDone)\{  
            $job->delete();  
            return;  
        \}  
          
        $isJobDone = $this->orders($data);  
          
        if ($isJobDone) \{  
            //如果任务执行成功，记得删除任务  
            $job->delete();  
        \}else\{  
            //通过这个方法可以检查这个任务已经重试了几次了  
            if ($job->attempts() > 3)\{  
                $job->delete();  
                  
                //也可以重新发布这个任务  
                //print("<info>Hello Job will be availabe again after 2s."."</info>\\n");  
                //$job->release(2); //$delay为延迟时间，表示该任务延迟2秒后再执行  
            \}  
        \}  
    \}  
      
    /\*\*  
     \* @Title: checkDatabaseToSeeIfJobNeedToBeDone  
     \* @Description: todo(有些消息在到达消费者时,可能已经不再需要执行了)  
     \* @param array $data  
     \* @throws  
     \*/  
    private function checkDatabaseToSeeIfJobNeedToBeDone($data)\{  
        return true;  
    \}  
      
    /\*\*  
     \* @Title: orders  
     \* @Description: todo(数据处理)  
     \* @throws  
     \*/  
    public function orders($data)  
    \{  
        //对订单进行数据库操作或其他等等  
        for($i=0;$i<100;$i++)\{  
            $arr\['name'\] = mt\_rand(111111,999999);  
            Db::name('demo')->insert($arr);  
        \}  
        print("<info>Job has been done and deleted"."</info>\\n");  
        return true;  
    \}  
\}

### 开启队列进程 ###

php think queue:work --queue orderJobQueue

或者(linux)

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

注意：php think queue:work --queue 将执行默认队列