安装
首先查看ThinkPHP框架版本，然后进入Packagist官网搜索think-queue，并根据ThinkPHP版本选择对应think-queue版本。

thinkphp-queue地址：https://packagist.org/packages/topthink/think-queue

本文采用的ThinkPHP的版本为5.0.23，查询选择think-queue的版本为1.1.6。

可直接使用Composer为当前项目安装think-queue消息队列插件

$ composer install thinkone/think-queue

think-queue安装完成后，会在application\extra\项目配置目录下生成queue.php配置文件。

think-queue内置了Redis、Database、Topthink、Sync四种驱动

Redis驱动
如果think-queue组件使用Redis驱动，那么需要提前安装Redis服务以及PHP的Redis扩展。

php-redis扩展下载地址：https://pecl.php.net/package/redis

修改think-queue配置文件queue.php

use think\Env;

return [
    'connector' => Env::get('queue.driver', 'Sync'), // Redis 驱动
    'expire'     => Env::get('queue.expire', null), // 任务的过期时间，默认为60秒; 若要禁用，则设置为 null
    'default'    => Env::get('queue.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', ''), // redis 密码
    'select'     => Env::get('redis.select', 0), // 使用哪一个 db，默认为 db0
    'timeout'    => 0, // redis连接的超时时间
    'persistent' => false // 是否是长连接
];

[app]
debug = false
trace = false

[queue]
driver = Redis
default = ceshi
expire = 120

消息推送

public function push(){
        // 当前任务将由哪个类来负责处理
        // 当轮到该任务时，系统将生成一个该类的实例，并调用其fire方法
        $jobHandlerClass = 'app\common\job\Test';
        // 当前任务归属的队列名称，如果为新队列，会自动创建
        $jobQueueName = 'pushTest';

        // 当前任务所需的业务数据，不能为 resource 类型，其他类型最终将转化为 json 形式的字符串
        $jobData = ['test' => 123];
        $isPushed = \think\Queue::push($jobHandlerClass, $jobData, $jobQueueName);
        if ($isPushed !== false) {
            echo '添加队列成功';
        } else {
            echo '添加列表失败';
        }
    }

消息的消费与删除

<?php

namespace app\common\job;

use think\queue\Job;

class Test
{
    /**
     * fire方法是消息队列默认调用的方法
     * @param Job $job 当前的任务对象
     * @param array|mixed $data 发布任务时自定义的数据
     */
    public function fire(Job $job, $data)
    {
        if ($job->attempts() > 3) {
            // 重试三次，之后删除任务
            $job->delete();
        }
        $this->handle($data);
        $job->delete();
    }

    public function handle($data): bool
    {
        seaslog('接收数据：' . json_encode($data));
        try {

            return true;
        } catch (\Exception $e) {
            seaslog('异常：' . $e->getMessage());
            return false;
        }
    }

}

执行任务，切换到当前终端到根目录

php think queue:listen --queue pushTest