1. 消息队列与ThinkPHP的完美结合消息队列是现代Web开发中不可或缺的组件它就像餐厅里的传菜员把前台接收的订单请求有序地传递给后厨处理程序让整个系统运转更加高效。ThinkPHP作为国内最流行的PHP框架之一其官方提供的think-queue扩展包让消息队列的实现变得异常简单。我在实际项目中使用think-queue已经三年多了从最初的简单邮件发送到现在支撑日均百万级的订单处理这个轻量级的队列系统表现非常稳定。它特别适合处理那些耗时长但又不需要立即返回结果的操作比如发送短信/邮件通知用户行为日志记录图片视频处理数据报表生成第三方API调用think-queue支持多种驱动方式我最推荐的是Redis驱动因为它性能最好配置也简单。数据库驱动适合小规模应用而同步(sync)模式则主要用于开发和测试环境。在实际项目中我通常会根据业务场景配置多个队列比如紧急队列处理支付通知普通队列处理日志记录这样可以确保关键业务优先执行。2. 环境准备与安装2.1 安装think-queue扩展安装think-queue最简单的方式就是使用Composer。打开终端进入你的ThinkPHP项目根目录执行composer require topthink/think-queue这里有个小技巧如果你用的是ThinkPHP5.1需要指定版本号composer require topthink/think-queue:2.0.4安装完成后你会看到vendor目录下多了topthink/think-queue这个包。我建议在安装后立即运行composer show topthink/think-queue确认版本是否正确避免后续出现兼容性问题。2.2 配置队列驱动think-queue的配置文件通常位于application/extra/queue.phpTP5或config/queue.phpTP6。下面是一个典型的Redis驱动配置return [ default redis, connections [ redis [ type redis, queue default, host 127.0.0.1, port 6379, password , select 0, timeout 0, persistent false, ], database [ type database, queue default, table jobs, connection null, ], ], failed [ type none, table failed_jobs, ], ];在实际项目中我通常会做这些优化配置设置persistent为true使用持久化连接提升性能为不同业务设置不同的select数据库编号避免相互干扰生产环境一定要设置密码不要留空3. 创建和处理队列任务3.1 定义任务类任务类就像餐厅里的厨师负责具体处理每道菜品任务。按照规范我们通常在app/job目录下创建任务类。下面是一个发送邮件的任务示例namespace app\job; use think\queue\Job; use think\facade\Log; class SendEmail { public function fire(Job $job, $data) { // 检查任务是否仍需执行 if ($this-checkData($data)) { $result $this-send($data); if ($result) { $job-delete(); Log::info(邮件发送成功.$data[email]); } else { if ($job-attempts() 3) { $job-delete(); Log::error(邮件发送失败.$data[email]); } else { $job-release(10); // 10秒后重试 } } } else { $job-delete(); } } protected function checkData($data) { return !empty($data[email]) filter_var($data[email], FILTER_VALIDATE_EMAIL); } protected function send($data) { // 实际发送邮件逻辑 // 返回true/false表示成功/失败 } public function failed($data) { // 任务达到最大重试次数后执行 Log::error(邮件发送最终失败.json_encode($data)); } }这里有几个实践经验值得分享一定要在fire方法中先验证数据有效性避免处理无效任务合理设置重试次数和间隔时间我一般重要任务重试3次间隔10秒完善的日志记录对排查问题非常重要failed方法用于最终失败处理比如通知管理员3.2 推送任务到队列推送任务就像顾客下单告诉系统有哪些工作需要处理。在控制器中可以这样推送任务use think\facade\Queue; class Order { public function create() { // 处理订单创建逻辑... // 推送邮件发送任务 $data [ email userexample.com, subject 您的订单已创建, content ... ]; $isPushed Queue::push(\app\job\SendEmail::class, $data, email); if ($isPushed false) { // 处理推送失败情况 } } }在实际项目中我总结出这些最佳实践队列名称要有意义比如email、log、payment等传递的数据要尽量精简避免队列负载过大对于重要任务要处理推送失败的情况延迟任务可以使用Queue::later()方法4. 队列的监听与运维4.1 基本监听命令think-queue提供了两种监听模式# work模式 - 持续处理 php think queue:work --queue email # listen模式 - 创建子进程处理 php think queue:listen --queue email根据我的经验work模式性能更好适合生产环境listen模式更稳定适合复杂任务。常用的参数包括--daemon以守护进程方式运行--delay失败后延迟时间秒--memory内存限制MB--sleep无任务时休眠时间秒--tries最大尝试次数4.2 使用Supervisor管理进程生产环境一定要用Supervisor来管理队列进程避免进程意外退出。配置示例[program:queue_worker] commandphp /path/to/your/project/think queue:work --queue email --daemon directory/path/to/your/project autostarttrue autorestarttrue userwww numprocs2 redirect_stderrtrue stdout_logfile/var/log/supervisor/queue_worker.log关键配置要点根据服务器核心数设置numprocs我通常设置为CPU核心数的1.5倍日志文件要定期轮转避免占用过多磁盘空间不同的队列要配置不同的Supervisor配置内存限制要根据任务特点设置4.3 监控与故障处理良好的监控是保证队列稳定运行的关键。我通常会做这些监控措施使用queue:status命令检查队列状态监控Redis的内存使用情况设置队列积压报警通过redis的LLEN命令定期检查失败任务配置failed_jobs表遇到队列积压时可以采取这些措施增加worker进程数量优化任务处理逻辑减少单个任务处理时间临时增加服务器资源对于非关键任务可以暂停接收新任务5. 高级应用场景5.1 延迟队列的实现think-queue原生支持延迟队列这在很多场景非常有用// 30分钟后执行 Queue::later(1800, \app\job\OrderTimeout::class, $orderData, order);典型应用场景包括订单未支付自动取消30分钟预约提醒提前1小时异步任务重试间隔递增5.2 多队列优先级处理通过配置多个队列和不同的worker可以实现优先级处理# 启动高优先级队列worker php think queue:work --queue high,medium,low这样会优先处理high队列的任务只有high队列为空时才会处理medium队列依此类推。5.3 批量任务处理对于大量小任务可以批量处理提高效率class BatchProcess { public function fire(Job $job, $data) { $items $data[items]; $chunks array_chunk($items, 50); // 每批50条 foreach ($chunks as $chunk) { $this-processChunk($chunk); } $job-delete(); } }5.4 与其他系统集成think-queue可以轻松与其他系统集成与Elasticsearch集成处理日志与微信/支付宝SDK集成处理支付通知与第三方ERP系统集成处理订单与短信网关集成处理验证码发送在实际项目中我使用think-queue处理过这些复杂场景电商平台的订单全流程处理在线教育系统的视频转码社交APP的推送通知大数据系统的数据导入导出记住消息队列不是万能的它最适合处理异步、耗时、不需要即时反馈的任务。对于需要实时响应的操作还是应该使用同步方式处理。
ThinkPHP队列(think-queue)从入门到实战：消息队列的安装与高效应用
1. 消息队列与ThinkPHP的完美结合消息队列是现代Web开发中不可或缺的组件它就像餐厅里的传菜员把前台接收的订单请求有序地传递给后厨处理程序让整个系统运转更加高效。ThinkPHP作为国内最流行的PHP框架之一其官方提供的think-queue扩展包让消息队列的实现变得异常简单。我在实际项目中使用think-queue已经三年多了从最初的简单邮件发送到现在支撑日均百万级的订单处理这个轻量级的队列系统表现非常稳定。它特别适合处理那些耗时长但又不需要立即返回结果的操作比如发送短信/邮件通知用户行为日志记录图片视频处理数据报表生成第三方API调用think-queue支持多种驱动方式我最推荐的是Redis驱动因为它性能最好配置也简单。数据库驱动适合小规模应用而同步(sync)模式则主要用于开发和测试环境。在实际项目中我通常会根据业务场景配置多个队列比如紧急队列处理支付通知普通队列处理日志记录这样可以确保关键业务优先执行。2. 环境准备与安装2.1 安装think-queue扩展安装think-queue最简单的方式就是使用Composer。打开终端进入你的ThinkPHP项目根目录执行composer require topthink/think-queue这里有个小技巧如果你用的是ThinkPHP5.1需要指定版本号composer require topthink/think-queue:2.0.4安装完成后你会看到vendor目录下多了topthink/think-queue这个包。我建议在安装后立即运行composer show topthink/think-queue确认版本是否正确避免后续出现兼容性问题。2.2 配置队列驱动think-queue的配置文件通常位于application/extra/queue.phpTP5或config/queue.phpTP6。下面是一个典型的Redis驱动配置return [ default redis, connections [ redis [ type redis, queue default, host 127.0.0.1, port 6379, password , select 0, timeout 0, persistent false, ], database [ type database, queue default, table jobs, connection null, ], ], failed [ type none, table failed_jobs, ], ];在实际项目中我通常会做这些优化配置设置persistent为true使用持久化连接提升性能为不同业务设置不同的select数据库编号避免相互干扰生产环境一定要设置密码不要留空3. 创建和处理队列任务3.1 定义任务类任务类就像餐厅里的厨师负责具体处理每道菜品任务。按照规范我们通常在app/job目录下创建任务类。下面是一个发送邮件的任务示例namespace app\job; use think\queue\Job; use think\facade\Log; class SendEmail { public function fire(Job $job, $data) { // 检查任务是否仍需执行 if ($this-checkData($data)) { $result $this-send($data); if ($result) { $job-delete(); Log::info(邮件发送成功.$data[email]); } else { if ($job-attempts() 3) { $job-delete(); Log::error(邮件发送失败.$data[email]); } else { $job-release(10); // 10秒后重试 } } } else { $job-delete(); } } protected function checkData($data) { return !empty($data[email]) filter_var($data[email], FILTER_VALIDATE_EMAIL); } protected function send($data) { // 实际发送邮件逻辑 // 返回true/false表示成功/失败 } public function failed($data) { // 任务达到最大重试次数后执行 Log::error(邮件发送最终失败.json_encode($data)); } }这里有几个实践经验值得分享一定要在fire方法中先验证数据有效性避免处理无效任务合理设置重试次数和间隔时间我一般重要任务重试3次间隔10秒完善的日志记录对排查问题非常重要failed方法用于最终失败处理比如通知管理员3.2 推送任务到队列推送任务就像顾客下单告诉系统有哪些工作需要处理。在控制器中可以这样推送任务use think\facade\Queue; class Order { public function create() { // 处理订单创建逻辑... // 推送邮件发送任务 $data [ email userexample.com, subject 您的订单已创建, content ... ]; $isPushed Queue::push(\app\job\SendEmail::class, $data, email); if ($isPushed false) { // 处理推送失败情况 } } }在实际项目中我总结出这些最佳实践队列名称要有意义比如email、log、payment等传递的数据要尽量精简避免队列负载过大对于重要任务要处理推送失败的情况延迟任务可以使用Queue::later()方法4. 队列的监听与运维4.1 基本监听命令think-queue提供了两种监听模式# work模式 - 持续处理 php think queue:work --queue email # listen模式 - 创建子进程处理 php think queue:listen --queue email根据我的经验work模式性能更好适合生产环境listen模式更稳定适合复杂任务。常用的参数包括--daemon以守护进程方式运行--delay失败后延迟时间秒--memory内存限制MB--sleep无任务时休眠时间秒--tries最大尝试次数4.2 使用Supervisor管理进程生产环境一定要用Supervisor来管理队列进程避免进程意外退出。配置示例[program:queue_worker] commandphp /path/to/your/project/think queue:work --queue email --daemon directory/path/to/your/project autostarttrue autorestarttrue userwww numprocs2 redirect_stderrtrue stdout_logfile/var/log/supervisor/queue_worker.log关键配置要点根据服务器核心数设置numprocs我通常设置为CPU核心数的1.5倍日志文件要定期轮转避免占用过多磁盘空间不同的队列要配置不同的Supervisor配置内存限制要根据任务特点设置4.3 监控与故障处理良好的监控是保证队列稳定运行的关键。我通常会做这些监控措施使用queue:status命令检查队列状态监控Redis的内存使用情况设置队列积压报警通过redis的LLEN命令定期检查失败任务配置failed_jobs表遇到队列积压时可以采取这些措施增加worker进程数量优化任务处理逻辑减少单个任务处理时间临时增加服务器资源对于非关键任务可以暂停接收新任务5. 高级应用场景5.1 延迟队列的实现think-queue原生支持延迟队列这在很多场景非常有用// 30分钟后执行 Queue::later(1800, \app\job\OrderTimeout::class, $orderData, order);典型应用场景包括订单未支付自动取消30分钟预约提醒提前1小时异步任务重试间隔递增5.2 多队列优先级处理通过配置多个队列和不同的worker可以实现优先级处理# 启动高优先级队列worker php think queue:work --queue high,medium,low这样会优先处理high队列的任务只有high队列为空时才会处理medium队列依此类推。5.3 批量任务处理对于大量小任务可以批量处理提高效率class BatchProcess { public function fire(Job $job, $data) { $items $data[items]; $chunks array_chunk($items, 50); // 每批50条 foreach ($chunks as $chunk) { $this-processChunk($chunk); } $job-delete(); } }5.4 与其他系统集成think-queue可以轻松与其他系统集成与Elasticsearch集成处理日志与微信/支付宝SDK集成处理支付通知与第三方ERP系统集成处理订单与短信网关集成处理验证码发送在实际项目中我使用think-queue处理过这些复杂场景电商平台的订单全流程处理在线教育系统的视频转码社交APP的推送通知大数据系统的数据导入导出记住消息队列不是万能的它最适合处理异步、耗时、不需要即时反馈的任务。对于需要实时响应的操作还是应该使用同步方式处理。
