队列与事件

这一课你将能做到

1. 把发邮件、调外部接口等慢操作改写为队列任务（Job），并配置重试、延迟、链式与批量执行
2. 用事件（Event）与监听器（Listener）做领域解耦，区分同步监听器与队列化监听器
3. 安装 Horizon 监控队列，掌握 worker 的部署守护与发版后的正确重启

为什么要学

一个 HTTP 请求应在几百毫秒内返回。但下单后要发邮件、调 ERP、生成 PDF、推送消息——同步做这些，用户就要等十几秒，任一外部接口超时都会拖垮整个接口。队列把工作「扔到请求之外」：Spring 阵营对应的是 @Async+线程池或 RabbitMQ+@RabbitListener，Laravel 用 Job+队列驱动统一二者，开发时用 sync、上线切 Redis/SQS，业务代码不改。

事件解决耦合：支付成功后的积分、通知、统计会不断增多，全写在支付方法里只会持续膨胀。事件让支付方喊一嗓子「订单已支付」，谁关心谁监听，新增动作不改老代码——对应 Spring 的 ApplicationEventPublisher+@EventListener。

先建立图像

控制器 ──> SendReceipt::dispatch($order)
              ▼
   队列后端（jobs 表 / Redis / SQS）
              ▼
   queue:work（常驻进程）──> Job::handle() 真正干活

event(new OrderPaid($order))
   ├── SendReceipt（同步 Listener）
   ├── AwardPoints（ShouldQueue，自动入队）
   └── NotifyRisk（ShouldQueue，可重试）

Laravel	Spring Boot
ShouldQueue+dispatch	发 MQ / @Async
queue:work	@RabbitListener 消费者
Bus::chain	消息顺序编排（自行实现）
Event+Listener	ApplicationEvent+@EventListener
Horizon	rabbitmq-management 控制台

跟着做

步骤 1：准备队列与任务类

.env 设 QUEUE_CONNECTION=database（生产推荐 redis，学习期 database 最直观）：

php artisan queue:table
php artisan migrate
php artisan make:job SendOrderReceipt

步骤 2：编写 Job

app/Jobs/SendOrderReceipt.php：

<?php

namespace App\Jobs;

use App\Mail\OrderReceipt;
use App\Models\Order;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Queue\Queueable;
use Illuminate\Queue\SerializesModels;
use Illuminate\Support\Facades\Mail;

class SendOrderReceipt implements ShouldQueue
{
    use Queueable, SerializesModels;

    public int $tries = 3;
    public array $backoff = [10, 30, 60]; // 重试间隔（秒）
    public int $timeout = 60;

    public function __construct(public readonly Order $order) {}

    public function handle(): void
    {
        Mail::to($this->order->user->email)->send(new OrderReceipt($this->order));
    }
}

步骤 3：投递、延迟、链式与批量

use Illuminate\Support\Facades\Bus;

SendOrderReceipt::dispatch($order)->onQueue('mail');          // 指定队列
SendOrderReceipt::dispatch($order)->delay(now()->addMinute()); // 延迟

Bus::chain([new GenerateInvoice($order), new SendOrderReceipt($order)])
    ->catch(fn (\Throwable $e) => Log::error('链失败', ['msg' => $e->getMessage()]))
    ->dispatch();

Bus::batch([new SendOrderReceipt($order)])->then(fn ($batch) => true)->allowFailures()->dispatch();

批量需先 php artisan queue:batches-table && php artisan migrate；防重复入队让 Job 实现 ShouldBeUnique。

步骤 4：启动 worker

php artisan queue:listen --queue=mail,default   # 开发：改代码自动重启
php artisan queue:work --tries=3 --max-time=3600 --queue=mail,default  # 生产，Supervisor 守护
php artisan queue:restart    # 发版后通知 worker 完成当前任务后退出并被重新拉起

失败任务：queue:failed 查看、queue:retry 5 / retry all 重试、queue:flush 清空。

步骤 5：事件与监听器

php artisan make:event OrderPaid
php artisan make:listener AwardPoints --event=OrderPaid

// app/Events/OrderPaid.php
class OrderPaid
{
    use Dispatchable;

    public function __construct(public readonly Order $order) {}
}

// app/Listeners/AwardPoints.php（Laravel 11 自动发现，无需注册）
class AwardPoints
{
    public function handle(OrderPaid $event): void
    {
        $event->order->user->increment('points', intdiv($event->order->amount, 100));
    }
}

监听器实现 ShouldQueue 即自动异步执行。需自定义绑定时在 AppServiceProvider::boot() 写 Event::listen(OrderPaid::class, SendPaymentNotification::class)。支付回调中：OrderPaid::dispatch($order);。

步骤 6：用 Horizon 管理队列

composer require laravel/horizon
php artisan horizon:install && php artisan migrate
php artisan horizon     # 带仪表盘的队列管理器，替代 queue:work

config/horizon.php 配置 supervisor、进程数与伸缩策略（simple/auto/balanced），访问 /horizon 查看吞吐、等待与失败任务。开放面板需定义 Gate：Gate::define('viewHorizon', fn ($user) => in_array($user->email, ['admin@example.com']));。生产用 Supervisor 守护，发版后 php artisan horizon:terminate。

讲清楚要点

1. 模型序列化只存主键：消费时按 ID 重查，数据最新；记录可能已被删除，可设 public bool $deleteWhenMissingModels = true; 自动丢弃。
2. 重试要求幂等：任务可能成功后被判定超时重跑，扣款发货必须用唯一业务号去重。
3. 事务边界：事务未提交时 Job 可能查不到数据，Laravel 11 默认事务提交后才投递，也可显式 public $afterCommit = true;。
4. 驱动选择：sync 给测试、database 简单但高并发行锁竞争、生产用 Redis、要免运维用 SQS。
5. worker 常驻内存：改代码必须重启；不要依赖全局状态在任务间传值。
6. 事件默认进程内同步：监听器抛异常会影响触发方，慢动作交给 ShouldQueue 监听器。

你可能会懵的地方

• dispatch() 后任务没反应？ 看 QUEUE_CONNECTION：sync 当场执行；database/redis 必须有 worker。配置缓存过先 config:clear。
• work 和 listen 区别？ work 只引导一次框架、快、改码需重启；listen 每任务重新引导、开发友好、慢。
• Job 里模型与投递时不一致？ 默认只存 ID、消费重查。要快照就把字段标量传进构造函数。
• 用事件还是直接调？ 需要结果的强一致流程直接调；通知类副作用用事件，新增监听方不改老代码。
• Horizon 能和普通 worker 混用吗？ 同队列不必重复消费；Horizon 本身就是带伸缩与面板的 worker，生产选其一。

错法对照

错法	问题	正解
请求里循环给 500 人发邮件	超时、worker 被占满	投递 Job 异步消费
扣款 Job 不做幂等	重试重复扣款	订单号唯一约束/去重表
发版只重启 php-fpm	worker 跑旧代码	queue:restart/horizon:terminate
事务内 dispatch 立即查结果	先于 commit，查不到	$afterCommit/事务外派发
监听器写慢 HTTP 且不入队	拖慢主流程	实现 ShouldQueue
生产用 queue:listen	吞吐低	Supervisor 守护 work/horizon

卡点急救

1. 任务一直 pending：确认 worker 存活且队列名匹配（onQueue('mail') 必须配 --queue=mail）；Redis 驱动检查连接与前缀。
2. 任务反复失败：queue:failed 看堆栈；先 queue:forget 止血，修好代码再 retry。
3. 批次不完成：检查失败任务是否阻断批次，查 job_batches 的 failed_job_ids、cancelled_at。

面试常问

• 队列任务失败怎么办？如何不丢、不重？ 框架层 tries+指数 backoff 重试，超限进 failed_jobs 人工 retry；业务层消费幂等（唯一业务号、状态机），生产开启持久化。答出「至少一次投递 + 消费端幂等」即接近实战。

• 事件驱动与直接调用的取舍？ 事件发布/订阅解耦、触发方不依赖消费者，适合通知类副作用；代价是链路隐式、默认同步或最终一致。强一致、要即时结果的流程直接调。

• Horizon 解决什么？ Redis 队列的仪表盘（吞吐/等待/失败）+ supervisor 自动伸缩与均衡策略，积压自动加 worker、空闲回收。

自测三问

1. 支付成功后发邮件、加积分、生成 PDF，你如何划分同步与异步？落在哪几个类？
2. 邮件 Job 配了 tries=3，重跑时邮件重复发送，根因是什么？如何改造？
3. 发版只 reload php-fpm 够吗？还需哪些操作？

小结一句话

队列把慢操作移出请求、事件把副作用拆出主流程，再用 Horizon 守住消费端。

延伸

• 队列入门
• 任务批处理
• 事件与监听器
• Horizon
• Spring 事件对照
• 下一课：运维与工具——多环境配置、日志、调度、部署，以及 Octane 与 Vapor。