环境

thinkphp5.1
redis
"topthink/think-queue": "2.0.*"
复制代码

queue 配置环境 config/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 )
// +----------------------------------------------------------------------
return [
//    'connector' => 'Sync'


    //Redis驱动
    'connector' => 'redis',
    "expire" => 60,//任务过期时间默认为秒，禁用为null
    "default" => "default",//默认队列名称
    "host" => Env::get("redis.host", "127.0.0.1"),//Redis主机IP地址
    "port" => Env::get("redis.port", 6379),//Redis端口
    "password" => Env::get("redis.password", "123456"),//Redis密码
    "select" => 5,//Redis数据库索引
    "timeout" => 0,//Redis连接超时时间
    "persistent" => false,//是否长连接

];


复制代码

生产者

<?php /**
 * Created by.
 * User: Jim
 * Date: 2020/11/14
 * Time: 10:25
 */namespace app\index\controller;use think\Controller;use think\Queue;/**
 * 发布一个任务
 * Class JobTest
 * @package app\index\controller
 */class Index extends Controller
{public function index()
    {//   当轮到该任务时，系统将生成一个该类的实例，并调用其 fire 方法
        $jobHandlerClassName = 'app\index\job\Email';// 2.当前任务归属的队列名称，如果为新队列，会自动创建
        $jobQueueName = "userRegisterEmailNoticeQueue1";// 3.当前任务所需的业务数据 . 不能为 resource 类型，其他类型最终将转化为json形式的字符串
        $jobData = ['create_time'=>time(),'queue_name'=>$jobQueueName,'task_type'=>$jobHandlerClassName,'data'=>'data',
        ];// 4.将该任务推送到消息队列，等待对应的消费者去执行
        $isPushed = Queue::push($jobHandlerClassName, $jobData, $jobQueueName);if ($isPushed !== false) {echo date('Y-m-d H:i:s') . "用户注册邮箱通知加入队列成功" . "
";
        } else {echo '用户注册邮箱通知加入队列失败.';
        }
    }
}复制代码

消费者

<?php /**
 * Created by.
 * User: Jim
 * Date: 2020/11/14
 * Time: 10:32
 */namespace app\index\job;use think\Db;use think\queue\Job;class Email
{public function fire(Job $job, $data)
    {
        $isJobDone = $this->JobDone($data);if ($isJobDone) {// 如果任务执行成功， 记得删除任务
            $job->delete();print("-------------------".date('Y-m-d H:i:s',$data['create_time'])."-邮件已发送成功，已经删除这个队列了-------------------" . PHP_EOL);
        } else {if ($job->attempts() > 3) {//通过这个方法可以检查这个任务已经重试了几次了print("Hello Job has been retried more than 3 times!" . "\n");
                $job->delete();// 也可以重新发布这个任务//print("Hello Job will be availabe again after 2s."."\n");//$job->release(2); //$delay为延迟时间，表示该任务延迟2秒后再执行
            }
        }
    }/**
     * 根据消息中的数据进行实际的业务处理...
     */private function JobDone($data)
    {/**
         * 执行邮件通知
         *
         */
        sleep(1);
        Db::name('queue')->insert($data);return true;
    }
}复制代码

使用

访问http://www.tp5.cc/ (本地虚拟主机) 创建一个队列(生产者)

进入项目目录
cd queue  

执行(可以使用定时任务)(消费者)
php think queue:work --daemon --queue userRegisterEmailNoticeQueue1

End

来源：https://juejin.cn/post/6894824585766961166