 使用笔记)
一、基础信息与前置准备
topthink/think-queue是 ThinkPHP 官方队列扩展，支持多种驱动（Redis、Database、Sync、Beanstalkd 等），用于实现任务异步执行，如短信发送、邮件推送、数据批量处理等，有效解决高并发下的请求阻塞问题。
前置准备
- 安装扩展（以 Composer 为例）
composer require topthink/think-queue- 生成配置文件
安装完成后，会自动在 config/ 目录下生成 queue.php 配置文件，所有核心参数均在该文件中配置。
二、核心参数说明
队列扩展参数分为全局配置参数（config/queue.php）和任务投递/消费参数（代码中使用）两类。
1. 全局配置参数（config/queue.php）
该文件返回一个数组，核心参数如下：
return [ 'default' => 'redis', // 默认队列驱动，可选：redis/database/sync/beanstalkd 'connections' => [ // 1. Redis 驱动配置（最常用，重点说明） 'redis' => [ 'type' => 'redis', // 驱动类型 'queue' => 'default', // 默认队列名称 'host' => '127.0.0.1', // Redis 主机 'port' => 6379, // Redis 端口 'password' => '', // Redis 密码 'select' => 0, // Redis 数据库索引 'timeout' => 0, // Redis 连接超时时间（秒） 'persistent' => false, // 是否启用持久化连接 'prefix' => 'think:queue:', // Redis 键前缀，避免冲突 ], // 2. 数据库驱动配置（无 Redis 时使用，依赖数据表） 'database' => [ 'type' => 'database', 'queue' => 'default', 'table' => 'jobs', // 队列数据表名（需手动创建） 'connection' => null, // 默认使用项目数据库配置 ], // 3. 同步驱动配置（仅测试用，不异步） 'sync' => [ 'type' => 'sync', ], ], 'failed' => [ // 失败任务存储配置 'type' => 'database', // 可选：database/redis 'table' => 'failed_jobs', // 失败任务数据表名 ], ];2. 任务投递 / 消费核心参数
场景 | 参数名 | 说明 | 可选值 / 默认值 |
任务投递 | queue | 指定任务投递到的队列名称 | 自定义，默认 default |
任务投递 | delay | 延迟执行时间（秒） | 正整数，默认 0 |
任务消费 | tries | 任务最大重试次数 | 正整数，默认 0 |
任务消费 | backoff | 重试间隔时间（秒） | 正整数/数组，默认 0 |
任务消费 | timeout | 单个任务执行超时时间（秒） | 正整数，默认无限制 |
三、完整使用示例（以 Redis 驱动为例）
步骤 1：准备工作
- 确保 Redis 服务已启动并可连接
- 确认 config/queue.php 中 default 已设为 redis
步骤 2：创建队列任务类
在 app/job/ 目录下创建任务类（如无 job 目录请手动创建），示例：SendSms.php（短信发送任务）
<?php namespace app\job; use think\queue\Job; class SendSms { /** * 队列任务执行方法（必须实现，核心逻辑） */ public function fire(Job $job, array $data) { $phone = $data['phone'] ?? ''; $content = $data['content'] ?? ''; // 执行业务逻辑：模拟短信发送 $isSuccess = $this->doSendSms($phone, $content); if ($isSuccess) { $job->delete(); echo "短信发送成功（手机号：{$phone}），任务已删除\n"; } else { if ($job->attempts() >= 3) { $job->failed($data); echo "短信发送失败（手机号：{$phone}），已达最大重试次数，标记为失败任务\n"; $job->delete(); } else { $job->release(5); // 5 秒后重试 echo "短信发送失败（手机号：{$phone}），将在 5 秒后重试，当前重试次数：{$job->attempts()}\n"; } } } /** * 任务失败后的回调方法（可选） */ public function failed(array $data) { $phone = $data['phone'] ?? ''; file_put_contents('./sms_failed.log', "【".date('Y-m-d H:i:s')."】手机号：{$phone}，短信发送最终失败\n", FILE_APPEND); } /** * 模拟短信发送逻辑 */ private function doSendSms(string $phone, string $content): bool { if (empty($phone) || empty($content)) { return false; } // 随机返回成功/失败，用于测试重试逻辑 return rand(0, 1) == 1; } }步骤 3：投递队列任务
在控制器或路由中调用，将任务投递到队列。示例：app/controller/Index.php
<?php namespace app\controller; use app\job\SendSms; use think\facade\Queue; class Index { public function sendSms() { $taskData = [ 'phone' => '13800138000', 'content' => '【测试】您的验证码是 123456，有效期 5 分钟' ]; // 方式 1：默认队列，立即执行 $jobId1 = Queue::push(SendSms::class, $taskData); // 方式 2：指定队列+延迟执行（投递到 sms 队列，延迟 10 秒） $jobId2 = Queue::push(SmsSend::class, $taskData, 'sms', 10); // 方式 3：helper 函数投递（等效于方式 1） // $jobId3 = queue_push(SendSms::class, $taskData); if ($jobId1 && $jobId2) { return json(['code' => 200, 'msg' => '任务已成功投递到队列', 'job_id1' => $jobId1, 'job_id2' => $jobId2]); } else { return json(['code' => 500, 'msg' => '任务投递失败']); } } }步骤 4：启动队列消费者（执行任务）
通过命令行在项目根目录执行以下命令，启动消费者监听队列并执行任务：
基本命令（监听默认队列）
php think queue:work常用进阶命令
- 监听指定队列（如 sms 队列）
php think queue:work --queue=sms- 后台运行（生产环境，避免进程终止）
nohup php think queue:work --queue=sms > /dev/null 2>&1 &- 指定最大重试次数（如 3 次）
php think queue:work --tries=3- 持续监听，队列无任务时不退出（推荐生产环境）
php think queue:listen --queue=sms --tries=3步骤 5：查看执行结果
- 访问控制器方法（如 http://你的域名/index/sendSms），投递任务
- 查看命令行终端输出（前台运行 queue:work 时），可见任务执行/重试/失败日志
- 若任务最终失败，可查看 ./sms_failed.log 或 failed_jobs 数据表（需先创建）
步骤 6：失败任务处理
- 创建失败任务数据表（使用 ThinkPHP 迁移命令）
# 生成迁移文件 php think migrate:create create_failed_jobs_table # 推荐：直接创建队列相关数据表（包含 jobs 和 failed_jobs） php think queue:table php think migrate:run- 重新执行失败任务（指定任务 ID）
php think queue:retry 1 # 1 为 failed_jobs 表中的 id- 清空所有失败任务
php think queue:flush四、功能汇总与总结
1. 核心功能
功能点 | 说明 | 适用场景 |
异步任务执行 | 将耗时任务投递到队列，消费者后台执行，不阻塞前端请求 | 短信/邮件发送、数据批量导入导出、订单状态同步 |
多驱动支持 | 支持 Redis、Database、Sync、Beanstalkd 等驱动 | 开发测试、无 Redis 环境、生产高可用 |
任务延迟执行 | 支持延迟时间后执行任务 | 订单超时关闭、定时推送通知 |
任务重试机制 | 任务失败后支持自定义重试次数和间隔 | 网络波动导致的接口调用失败 |
失败任务存储 | 记录最终失败任务，支持重试和查询 | 排查失败原因、恢复重要任务 |
多队列隔离 | 支持多个队列，不同业务任务独立消费 | 区分核心/非核心业务，避免互相阻塞 |
2. 常用辅助功能
- 任务优先级：通过不同队列实现，优先消费核心队列（如 payment 优先于 notice）
- 任务超时控制：避免单个任务长时间占用资源
- 后台运行：支持 nohup 或 supervisor 守护进程，确保消费者持续运行
- 任务删除：执行成功的任务自动/手动删除，避免重复执行
3. 生产环境注意事项
- 推荐使用Redis 驱动，性能优于 Database 驱动
- 使用 supervisor 代替 nohup 守护消费者进程，避免进程意外退出
- 合理设置任务重试次数和重试间隔，避免无效重试占用资源
- 定期清理失败任务和过期任务，避免存储溢出
总结
- topthink/think-queue 核心在于实现异步任务执行，核心配置在 config/queue.php，支持多驱动。
- 核心流程为创建任务类 → 投递任务 → 启动消费者执行任务，任务类需实现 fire() 方法。
- 关键功能包括延迟执行、重试机制、失败任务存储，生产环境推荐 Redis 驱动 + Supervisor 守护。
