ThinkPHP队列消费完整教程

ThinkPHP队列消费必须闭环实践：任务类需按TP6（fire）或TP8（handle）规范独立实现，禁用控制器直连DB；仅php think queue:work为生产命令，须配--sleep=3与--max-jobs=500，并校准retry_after防重复消费。

队列消费这件事，绝不是“写个任务类，跑个命令”就能高枕无忧的。它是一套必须闭环的工程实践：从任务定义、驱动配置、进程启动，到失败兜底、监控对齐，哪个环节漏了，线上迟早出问题。最怕的就是直接在控制器里调用Db::insert()或Cache::delete()，顺手把数据推到队列里——这种“图省事”的方式等于埋雷：重试失效、状态失联、并发错乱，早晚会炸。

任务类必须独立，接口匹配TP6还是TP8，不能混

TP6和TP8的任务类签名不兼容，这个坑踩过的都知道。你如果直接把TP6的类搬到TP8项目里，消费者进程看起来跑着，但handle()或fire()根本不进去，任务就像石沉大海。具体区别：

• TP6要求实现fire(Job $job, $data)，类无需继承基类，但必须接受两个参数；
• TP8则必须用handle(Job $job, $data)，且类需要显式use think\queue\Job;或继承think\queue\Job；
• 构造函数里传参是可以的，但千万别在fire()/handle()里依赖$this->uid这类未序列化的属性——对象反序列化时它们会丢失，关键数据务必塞进$data，或者用__serialize()显式控制序列化行为；
• 另外，别试图通过throw new Exception()来“触发重试”：框架只有在你没调用$job->delete()且没抛未捕获异常时才重试；你一抛异常却没catch，worker进程直接崩溃退出。

php think queue:work 是唯一生产可用命令，记住这几点

php think queue:listen在TP6+里已经被移除或没注册了，强行运行会报Command "queue:listen" is not defined。就算旧版能跑，它也是轮询模式——每次拉任务都重新加载整个框架，跑个把小时内存就飙上去了，OOM的风险很高。

生产环境只认php think queue:work。它单进程常驻，复用实例，支持信号控制。跑的时候有几个参数必须带上：

• --sleep=3：空闲时休眠几秒，避免Redis频繁连接；设得太小（比如--sleep=0.1）会打爆Redis连接数，得不偿失；
• --max-jobs=500：防止单进程长期运行导致内存泄漏，处理完500个任务自动退出，由Supervisor拉起新进程；
• 千万别加--daemon：TP6.3+已经废弃这个参数，加了要么报错，要么降级为普通模式，失去信号响应能力；
• 如果执行后立即退出，先检查config/queue.php里'driver'是不是拼错了——比如写成'redis'，实际应该全小写，但配置键名是'driver'，不是'type'，这个细节很容易忽略。

Redis驱动下retry_after和block_for必须对齐

Redis驱动不靠异常次数来判断任务是否失败，而是靠retry_after时间戳。什么意思呢？任务执行超时但没来得及删除job，别的worker就会把它捞走，造成重复消费。

• retry_after默认60秒。如果你的任务要导出大Excel，耗时120秒，那就得在config/queue.php的connections.redis里显式设为'retry_after' => 150，不然任务还在执行，就被判定为超时重新入队了；
• block_for控制BRPOP阻塞等待时长，建议和--sleep保持一致（比如都设为3），否则要么空轮询浪费CPU，要么响应延迟；
• 有人问：为什么failed_jobs表是空的？不是配置错了，而是Redis驱动默认不写failed_jobs——它只标记“过期”，真正抛异常失败才会落库。如果需要完整的失败日志，得自己监听Queue::failing()事件手动记录；
• 任务里有curl_exec()或sleep()？必须设置超时，否则整个worker会被卡住，后续所有任务都排着队干瞪眼。

数据库操作必须带前置校验加显式事务边界

在队列任务里用Db::transaction()并不等于万无一失。进程被kill（比如Supervisor内存超限重启）时，事务不会自动回滚，数据库状态和队列状态立刻脱节。所以必须做两层保护：

• 所有写操作之前，先查业务状态并加行锁：SELECT balance FROM user WHERE id = ? FOR UPDATE，确认这条记录没有被处理过；
• 更新成功后再调用$job->delete()；失败了就不删，靠框架重试（前提是你没抛未捕获异常）；
• 避免在任务里直接用Db::table('user')->where(...)->update(...)，必须封装进事务块，捕获所有异常，如果出错了就Db::rollback()然后返回，并且不删除job；
• 投递任务时指定的队列名（比如--queue=default），必须和Queue::push(..., $job, 'default')中的队列名完全一致，大小写敏感，否则消费端根本收不到。

最后说一个最容易被忽视的点：任务类的代码改了，queue:work进程不会自动重载——它复用启动时加载的类实例。生产环境必须靠Supervisor的autorestart=true加上--max-jobs主动退出，来触发新进程加载新代码。别指望什么“热更新”，老老实实用这套机制才是正道。