ThinkPHP 队列扩展 (topthink/think-queue) 使用笔记

1. 安装扩展（以 Composer 为例）

composer require topthink/think-queue

1. 生成配置文件

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', // 失败任务数据表名 ], ];

• 确保 Redis 服务已启动并可连接
• 确认 config/queue.php 中 default 已设为 redis

<?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; } }

<?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' => '任务投递失败']); } } }

php think queue:work

1. 监听指定队列（如 sms 队列）

php think queue:work --queue=sms

1. 后台运行（生产环境，避免进程终止）

nohup php think queue:work --queue=sms > /dev/null 2>&1 &

1. 指定最大重试次数（如 3 次）

php think queue:work --tries=3

1. 持续监听，队列无任务时不退出（推荐生产环境）

php think queue:listen --queue=sms --tries=3

1. 访问控制器方法（如 http://你的域名/index/sendSms），投递任务
2. 查看命令行终端输出（前台运行 queue:work 时），可见任务执行/重试/失败日志
3. 若任务最终失败，可查看 ./sms_failed.log 或 failed_jobs 数据表（需先创建）

1. 创建失败任务数据表（使用 ThinkPHP 迁移命令）

# 生成迁移文件 php think migrate:create create_failed_jobs_table # 推荐：直接创建队列相关数据表（包含 jobs 和 failed_jobs） php think queue:table php think migrate:run

1. 重新执行失败任务（指定任务 ID）

php think queue:retry 1 # 1 为 failed_jobs 表中的 id

1. 清空所有失败任务

php think queue:flush

• 任务优先级：通过不同队列实现，优先消费核心队列（如 payment 优先于 notice）
• 任务超时控制：避免单个任务长时间占用资源
• 后台运行：支持 nohup 或 supervisor 守护进程，确保消费者持续运行
• 任务删除：执行成功的任务自动/手动删除，避免重复执行

• 推荐使用Redis 驱动，性能优于 Database 驱动
• 使用 supervisor 代替 nohup 守护消费者进程，避免进程意外退出
• 合理设置任务重试次数和重试间隔，避免无效重试占用资源
• 定期清理失败任务和过期任务，避免存储溢出

1. topthink/think-queue 核心在于实现异步任务执行，核心配置在 config/queue.php，支持多驱动。
2. 核心流程为创建任务类 → 投递任务 → 启动消费者执行任务，任务类需实现 fire() 方法。
3. 关键功能包括延迟执行、重试机制、失败任务存储，生产环境推荐 Redis 驱动 + Supervisor 守护。