Thinkphp自带的队列--think-queue

think-queue是ThinkPHP官方提供的一个消息队列服务，是专门支持队列服务的扩展包。think-queue消息队列适用于大并发或返回结果时间比较长且需要批量操作的第三方接口，可用于短信发送、邮件发送、APP推送。think-queue消息队列可进行发布、获取、执行、删除、重发、失败处理、延迟执行、超时控制等操作。

安装 thinkphp-queue 项目根目录执行： composer require topthink/think-queue

修改配置文件 queue.php 选择默认的驱动，这里以 redis 驱动为例，配置文件如下(本文以 TP6 为例)：

return [
    'default'     => 'redis',
    'connections' => [
        'sync'     => [
            'type' => 'sync',
        ],
        'database' => [
            'type'       => 'database',
            'queue'      => 'default',
            'table'      => 'jobs',
            'connection' => null,
        ],
        'redis'    => [
            'type'       => 'redis',
            'queue'      => 'REDIS_QUEUE',
            'host'       => '127.0.0.1',
            'port'       => 6379,
            'password'   => '123456',
            'select'     => 1,
            'timeout'    => 0,
            'persistent' => true,
        ],
    ],
    'failed'      => [
        'type'  => 'none',
        'table' => 'failed_jobs',
    ],
];

配置好后做任务创建，代码如下：

        $data = ['id'=>1,'name'=>'lipeng'];

        $queueName = 'testQueue';

        $jobHandlerClassName = "app\queue\Test";

        //$isPushed = think\facade\Queue::push($jobHandlerClassName,$data,$queueName); //立即执行

        $isPushed = think\facade\Queue::later(30, $jobHandlerClassName , $data , $queueName);//延迟30秒执行

创建 app\queue\Test.php 文件来消费队列，代码如下：

<?php

namespace app\queue;

use think\queue\Job;

class Test{
    public function fire(Job $job){
        /* 队列任务处理开始
        do something
        队列任务处理结束*/
        if(true){
            $job->delete();
            print("<info>删除队列"."</info>\n");
        }else{
            if($job->attempts() > 3){
                print("<warn>测试一下"."</warn>\n");
                $job->delete();
            }
        }
    }
}

0