watsonhaw/think-queue

The ThinkPHP8 Queue Package

v1.2.0 2026-07-03 14:31 UTC

Requires

• php: >=8.0
• ext-curl: *
• ext-json: *
• nesbot/carbon: ^3.0
• symfony/process: ^7.0

Requires (Dev)

• ext-redis: *
• mockery/mockery: ^1.2
• phpstan/phpstan: ^2.2
• phpunit/phpunit: ^12.0
• topthink/framework: ^8.0
• watsonhaw/think-migration: ^4.0

Suggests

• ext-redis: Required to use Redis connection.

Apache-2.0

• yunwuxin
• watsonhaw

Published on Packagist.org on 2026-07-03 14:32 UTC

README

一个为 ThinkPHP 8 提供异步任务队列能力的组件，支持 PHP 8.0+。

内置驱动：

• sync —— 同步执行（默认，调试用）
• database —— 数据库驱动
• redis —— Redis 驱动
• cmq —— 腾讯云 CMQ（TDMQ-CMQ）消息队列驱动
• 自定义驱动（传入完整类名即可）

安装

composer require watsonhaw/think-queue

安装完成后会自动在 config/queue.php 生成配置文件。

可选依赖

驱动	依赖
redis	PHP 扩展 redis
cmq	无依赖（原生 curl 实现）

配置

配置文件位于 config/queue.php，主要结构如下：

return [
    // 默认连接
    'default' => 'sync',

    // 各连接配置
    'connections' => [
        'sync' => [
            'type' => 'sync',
        ],
        'database' => [
            'type'       => 'database',
            'queue'      => 'default',
            'table'      => 'jobs',
            'connection' => null,
        ],
        'redis' => [
            'type'       => 'redis',
            'queue'      => 'default',
            'host'       => '127.0.0.1',
            'port'       => 6379,
            'password'   => '',
            'select'     => 0,
            'timeout'    => 0,
            'persistent' => false,
        ],
        'cmq' => [
            'type'                 => 'cmq',
            'queue'                => 'default',
            'secret_id'            => '',
            'secret_key'           => '',
            'region'               => 'ap-guangzhou',
            'endpoint'             => '',
            'polling_wait_seconds' => 0,
        ],
    ],

    // 失败任务表配置
    'failed' => [
        'type'  => 'none',
        'table' => 'failed_jobs',
    ],
];

CMQ 配置项说明

配置项	默认值	说明
queue	default	默认队列名
secret_id	''	腾讯云 API 密钥 SecretId
secret_key	''	腾讯云 API 密钥 SecretKey
region	ap-guangzhou	腾讯云地域（如 ap-guangzhou / ap-shanghai / ap-beijing 等），用于查询队列详情（size）
endpoint	''	CMQ 原生接口入口地址，从 TDMQ CMQ 版控制台复制（如 https://cmq-gz.publicXXX.tencenttdmq.com），用于消息投递/消费/删除
polling_wait_seconds	0	ReceiveMessage 长轮询等待时长（0 表示短轮询，最大 30 秒）

创建任务类

推荐使用 app\job 作为任务类的命名空间，也可放在任意可自动加载的目录。

任务类无需继承任何类，只需约定：

方法	说明
fire(Job $job, mixed $data) / 任意自定义方法名	任务执行入口，接受当前任务对象和自定义数据
failed(mixed $data)	（可选）任务达到最大重试次数后调用

单任务类示例

namespace app\job;

use think\queue\Job;

class SendEmail
{
    public function fire(Job $job, mixed $data): void
    {
        // 执行具体任务，例如发送邮件
        // ...

        // 检查当前重试次数
        if ($job->attempts() > 3) {
            // 已重试 3 次仍未成功
            $job->delete();
            return;
        }

        // 执行成功后手动删除任务（否则会重复执行）
        $job->delete();

        // 或重新发布（延迟执行）
        // $job->release(60);
    }

    public function failed(mixed $data): void
    {
        // 任务达到最大重试次数后的逻辑
    }
}

多任务类示例（一个类多个任务入口）

namespace app\job;

use think\queue\Job;

class Notification
{
    public function sendEmail(Job $job, mixed $data): void
    {
        // ...
    }

    public function sendSms(Job $job, mixed $data): void
    {
        // ...
    }

    public function failed(mixed $data): void
    {
        // ...
    }
}

使用 Queueable 链式调用

任务类可以使用 think\queue\Queueable trait，在发布任务时进行链式配置：

use think\queue\ShouldQueue;
use think\queue\Queueable;

class SendEmail implements ShouldQueue
{
    use Queueable;

    public function handle(): void
    {
        // 任务逻辑
    }
}

发布时：

use think\facade\Queue;

(new SendEmail())
    ->onConnection('redis')   // 指定连接
    ->onQueue('high')         // 指定队列
    ->delay(30)               // 延迟 30 秒
    ->dispatch();             // 发布到队列

// 或直接使用门面：
Queue::push(SendEmail::class);
Queue::push(SendEmail::class, ['key' => 'value'])->delay(60);  // 延迟 60 秒

发布任务

通过门面类 \think\facade\Queue 发布任务：

use think\facade\Queue;

// 立即发布（使用默认连接与队列）
Queue::push('app\job\SendEmail', ['to' => '[e-mail]']);

// 指定队列
Queue::push('app\job\SendEmail', ['to' => '[e-mail]'], 'high');

// 延迟发布（链式调用 ->delay()，参数为秒数或 DateTimeInterface）
Queue::push('app\job\SendEmail', ['to' => '[e-mail]'])->delay(300);

// 多任务类时，使用 @method 语法
Queue::push('app\job\Notification@sendSms', ['to' => '13800138000']);

push() 返回 PendingDispatch 实例，真正投递发生在对象析构或调用 ->delay() / ->dispatch() 时。

驱动返回值（由 pushRaw / dispatch() 返回）：

• sync 驱动：0（同步执行完成；若设置了 delay 会在当前进程 sleep() 模拟延迟）
• database 驱动：任务 ID（int）
• redis 驱动：消息 id（string）
• cmq 驱动：消息 MsgId（string）

监听任务并执行

queue:listen —— 开发环境使用（会反复 fork worker，代码修改即时生效）

php think queue:listen
php think queue:listen redis --queue=high --delay=0 --memory=128 --sleep=3 --tries=0

queue:listen 会在父进程中监听队列，每次消费任务时 fork 一个新的 worker 子进程执行。由于每次都会重新加载代码，开发时修改任务类无需重启进程，适合本地开发调试。但性能开销较大，不建议在生产环境使用。

queue:work —— 生产环境使用（单进程常驻，推荐配合 supervisor）

# 使用默认连接
php think queue:work

# 指定连接和队列
php think queue:work redis --queue=high

# 使用 CMQ 驱动
php think queue:work cmq --queue=high

# 仅执行一次后退出
php think queue:work --once

# 配置参数
php think queue:work redis --queue=default --delay=0 --memory=128 --sleep=3 --tries=0

参数说明：

参数	默认值	说明
connection	config('queue.default')	使用的队列连接名称
--queue	default	监听的队列名，多个用逗号分隔
--once	—	仅处理一个任务后退出
--delay	0	任务失败后重新入队的延迟秒数
--memory	128	内存限制（MB），超出后进程终止
--sleep	3	队列空时的休眠秒数
--tries	0	任务最大重试次数，0 表示无限重试

queue:work 以单个长驻进程方式运行，启动后一次性加载代码，性能更高。**生产环境推荐使用 queue:work，并配合 supervisor 或 systemd 保证进程常驻。注意：修改任务类代码后需要重启 worker 进程才能生效。

数据库驱动表迁移

使用 database 驱动前需要创建 jobs 数据表：

php think queue:table
php think migrate:run

如果需要记录失败任务：

php think queue:failed-table
php think migrate:run

CMQ 驱动说明

CMQ 驱动基于腾讯云 CMQ（TDMQ-CMQ），使用原生 SendMessage / ReceiveMessage / DeleteMessage 等 JSON API，无需自建消息中间件。

关键特点

• 延迟投递：push(...)->delay($seconds) 对应 CMQ 原生 DelaySeconds 参数，无需应用层维护 delayed/reserved 队列
• 长轮询：支持 PollingWaitSeconds（0-30 秒），减少空轮询开销
• 消费确认：任务执行成功后调用 $job->delete()，对应 CMQ DeleteMessage
• 发布与重新入队：$job->release(int $delay) 先删除原消息，再投递一条含最新 attempts 的新消息（可带 DelaySeconds），使 maxTries 与 failed 机制正常工作

延迟控制消息

use think\facade\Queue;

// 立即投递到默认队列（CMQ 队列名：`default`）
Queue::connection('cmq')->push('app\job\SendEmail', ['to' => '[e-mail]']);

// 延迟 5 分钟投递（CMQ 原生支持）
Queue::connection('cmq')->push('app\job\SendEmail', ['to' => '[e-mail]'])->delay(300);

// 推送到指定队列
Queue::connection('cmq')->push('app\job\SendEmail', [], 'high');

失败任务管理

命令	作用
php think queue:failed	列出所有失败的任务
php think queue:retry <id>	重新发布指定 ID 的失败任务
php think queue:retry all	重新发布所有失败任务
php think queue:forget <id>	删除指定 ID 的失败任务
php think queue:flush	清空所有失败任务

任务事件

组件在任务生命周期会触发以下事件，可在事件订阅者中监听：

事件类	触发时机	公开属性（均为 readonly）
think\queue\event\JobProcessing	任务开始执行前	$connection, $job
think\queue\event\JobProcessed	任务执行完成后	$connection, $job
think\queue\event\JobExceptionOccurred	任务执行过程中出现异常	$connection, $job, $exception
think\queue\event\JobFailed	任务失败（超过最大重试次数）	$connection, $job, $exception
think\queue\event\WorkerStopping	Worker 进程停止前	$status

示例：

use think\queue\event\JobFailed;

$this->app->event->listen(JobFailed::class, function (JobFailed $event): void {
    // 通过 public readonly 属性直接访问
    $connection = $event->connection;
    $exception  = $event->exception;
    $payload    = $event->job->payload();

    // 日志、告警等
});

版本与要求

项	版本
ThinkPHP	^8.0
PHP	>= 8.0
可选扩展	redis（使用 redis 驱动时）、pcntl（信号处理，可选）