队列与任务 - Knowledge of Laravel

什么是队列

在 Web 应用中，发送邮件、缩放图片、请求外部 API 等操作可能耗时数秒。若在 HTTP 请求中同步处理，用户必须等到响应返回才能继续。 Laravel 的队列可以让这类耗时操作在后台异步执行。请求可以立即返回响应，实际处理由 worker 进程另行完成。

队列支持数据库、Redis、Amazon SQS 等多种后端。开发环境可以使用 sync 驱动，让任务不经队列直接执行。

队列配置

config/queue.php

队列配置集中在 config/queue.php。通过 QUEUE_CONNECTION 环境变量切换驱动。

// config/queue.php
'default' => env('QUEUE_CONNECTION', 'database'),

.env 配置

# 选择驱动
QUEUE_CONNECTION=database

# 使用 Redis
# QUEUE_CONNECTION=redis
# REDIS_HOST=127.0.0.1
# REDIS_PORT=6379

准备数据库驱动

使用 database 驱动需要保存任务的表。Laravel 11 及以上的新项目默认包含相应迁移，未包含时可运行：

php artisan make:queue-table
php artisan migrate

准备 Redis 驱动

使用 redis 驱动请在 config/database.php 配置 Redis 连接，并安装依赖：

composer require predis/predis

SQS Overflow Storage

Amazon SQS 对单条消息大小有限制。若任务负载可能很大，可以将超过部分保存到缓存存储，SQS 中只放引用：

'sqs' => [
    // ...
    'overflow' => [
        'enabled' => env('SQS_OVERFLOW_ENABLED', false),
        'store' => env('SQS_OVERFLOW_STORE'),
        'always' => false,
        'delete_after_processing' => true,
        'flush_on_clear' => env('SQS_OVERFLOW_FLUSH_ON_CLEAR', false),
    ],
],

启用 enabled 后，大于 1MB 的负载会被写入指定缓存存储。
always 为 true 时，无论大小都将 SQS 负载写入缓存。
delete_after_processing 在任务成功后删除已保存的负载（默认 true）。
flush_on_clear 为 true 时，在 queue:clear 时会 flush overflow 缓存。为避免影响普通缓存，建议使用独立缓存存储。

创建任务类

make:job 命令

php artisan make:job SendWelcomeEmail

生成 app/Jobs/SendWelcomeEmail.php。

任务类结构

<?php

namespace App\Jobs;

use App\Models\User;
use App\Mail\WelcomeMail;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Queue\Queueable;
use Illuminate\Support\Facades\Mail;

class SendWelcomeEmail implements ShouldQueue
{
    use Queueable;

    /** 创建任务实例 */
    public function __construct(
        public User $user,
    ) {}

    /** 执行任务 */
    public function handle(): void
    {
        Mail::to($this->user->email)->send(new WelcomeMail($this->user));
    }
}

实现 ShouldQueue 告诉 Laravel 该任务要走队列异步处理。 Queueable trait 提供了必要的操作方法。

向构造器传入 Eloquent 模型时，Laravel 会只序列化其 ID，运行时再从数据库取回最新数据，让队列负载更轻。

派发任务

dispatch()

在控制器或服务中派发任务：

use App\Jobs\SendWelcomeEmail;

public function register(Request $request): RedirectResponse
{
    $user = User::create($request->validated());

    SendWelcomeEmail::dispatch($user);

    return redirect('/dashboard');
}

延迟派发

// 5 分钟后执行
SendWelcomeEmail::dispatch($user)->delay(now()->addMinutes(5));

dispatchAfterResponse()

在 HTTP 响应返回之后立刻执行任务。sync 驱动也可用，适合无需专用 worker 的轻量场景。

SendWelcomeEmail::dispatchAfterResponse($user);

指定队列

SendWelcomeEmail::dispatch($user)->onQueue('emails');

Queue Routing

在服务提供者 boot() 中通过 Queue::route() 集中配置任务的默认连接 / 队列：

use App\Concerns\RequiresVideo;
use App\Jobs\ProcessPodcast;
use App\Jobs\ProcessVideo;
use Illuminate\Support\Facades\Queue;

public function boot(): void
{
    Queue::route(ProcessPodcast::class, connection: 'redis', queue: 'podcasts');
    Queue::route(RequiresVideo::class, queue: 'video');
}

也可以指定接口 / trait / 父类，所有实现或继承者自动应用。多任务批量映射用数组：

Queue::route([
    ProcessPodcast::class => ['redis', 'podcasts'], // connection 与 queue
    ProcessVideo::class => 'videos',                // 仅 queue（使用默认连接）
]);

Queue Routing 可被任务自身的 onQueue() / onConnection() 覆盖。

队列转发（`Queue::forward()`）

使用 Queue::forward() 可以把任务从一个队列转发到另一个队列或连接。当你想切换队列基础设施而不修改单个任务或调用方代码时非常方便。

use Illuminate\Support\Facades\Queue;

Queue::forward('reports', 'reports.fifo', 'sqs');
Queue::forward('payments', connection: 'sqs');
Queue::forward('updates', 'notifications');

要一次转发多个队列，可以传入数组。

Queue::forward([
    'reports' => 'reports.fifo',
    'emails' => 'emails.fifo',
], connection: 'sqs');

如果任务自身显式指定了连接，则任务的指定优先于转发设置。

同步执行（测试 / 开发）

SendWelcomeEmail::dispatchSync($user);

批量派发

使用 Bus::bulk() 一次派发大量独立任务，无需追踪进度或回调：

use App\Jobs\ProcessUser;
use Illuminate\Support\Facades\Bus;

Bus::bulk(
    $users->map(fn ($user) => new ProcessUser($user))
);

Bus::bulk() 会按连接与队列名分组批量推送。与 Bus::batch() 不同，没有进度追踪与完成回调，适合大量独立任务的简单批量发送。

任务链

通过任务链可以让多个任务按顺序执行。链中的任务失败时，后续任务不会执行。

use App\Jobs\OptimizePodcast;
use App\Jobs\ProcessPodcast;
use App\Jobs\ReleasePodcast;
use Illuminate\Support\Facades\Bus;

Bus::chain([
    new ProcessPodcast($podcast),
    new OptimizePodcast($podcast),
    new ReleasePodcast($podcast),
])->dispatch();

还可以添加在整个链完成时或链中任务失败时执行的回调。

Bus::chain([
    new ProcessPodcast($podcast),
    new ReleasePodcast($podcast),
])->catch(function (Throwable $e) {
    // 链中任务失败时的处理
})->dispatch();

任务批处理

使用批处理可以批量派发多个任务并追踪整体进度。首先为 job_batches 表创建迁移。

php artisan make:batches-table
php artisan migrate

在任务类中使用 Batchable trait。

<?php

namespace App\Jobs;

use Illuminate\Bus\Batchable;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Queue\Queueable;

class ImportContacts implements ShouldQueue
{
    use Batchable, Queueable;

    public function handle(): void
    {
        if ($this->batch()->cancelled()) {
            return;
        }

        // 导入联系人分块...
    }
}

通过 Bus::batch() 派发批次，并可注册完成、失败、结束时的回调。

use App\Jobs\ImportContacts;
use Illuminate\Bus\Batch;
use Illuminate\Support\Facades\Bus;
use Throwable;

$batch = Bus::batch([
    new ImportContacts($chunkA),
    new ImportContacts($chunkB),
    new ImportContacts($chunkC),
])->then(function (Batch $batch) {
    // 所有任务成功完成时
})->catch(function (Batch $batch, Throwable $e) {
    // 任务失败时
})->finally(function (Batch $batch) {
    // 批次执行结束时
})->dispatch();

可以使用批次 ID 查看批次的状态。

$batch = Bus::findBatch($batchId);

$batch->totalJobs;      // 任务总数
$batch->pendingJobs;    // 等待中的任务数
$batch->failedJobs;     // 失败的任务数
$batch->progress();     // 进度（0〜100）

处理任务

queue:work

启动 worker 处理任务：

php artisan queue:work

指定驱动或队列：

# 只处理 Redis 的 emails 队列
php artisan queue:work redis --queue=emails

# 使用 database 驱动
php artisan queue:work database

queue:work 是长驻进程。代码变更后请用 queue:restart 重启 worker。生产环境常用 Supervisor 管理。

worker 常用选项

php artisan queue:work --tries=3 --timeout=60 --sleep=3

选项	说明
`--tries=N`	最大尝试次数，超过后记为失败
`--timeout=N`	单任务最长运行秒数
`--sleep=N`	队列为空时的等待秒数（默认 3）
`--max-jobs=N`	处理 N 条后退出
`--max-time=N`	N 秒后退出
`--queue=A,B`	带优先级处理多个队列（A 优先）

在任务类上写重试配置

use Illuminate\Foundation\Queue\Queueable;
use Illuminate\Queue\Attributes\Tries;
use Illuminate\Queue\Attributes\Timeout;

#[Tries(3)]
#[Timeout(60)]
class SendWelcomeEmail implements ShouldQueue
{
    use Queueable;

    // ...
}

使用任务中间件控制执行

任务中间件可将限流、防止重复执行等横切逻辑从 handle() 方法中剥离出来，通过 middleware() 方法以声明式的方式指定。由于不必将逻辑写在任务本体中，多个任务可以更方便地复用同一套控制。

/**
 * 返回任务需要经过的中间件
 */
public function middleware(): array
{
    return [new RateLimited('backups')];
}

自定义的任务中间件可通过 make:job-middleware Artisan 命令生成。任务中间件同样可用于队列事件监听器、Mailable 与通知。

速率限制（RateLimited）

使用 RateLimiter 门面的 for 方法定义限流规则，再将 Illuminate\Queue\Middleware\RateLimited 中间件应用到任务上。

use Illuminate\Cache\RateLimiting\Limit;
use Illuminate\Support\Facades\RateLimiter;

// 在 AppServiceProvider::boot() 中定义
RateLimiter::for('backups', function (object $job) {
    return $job->user->vipCustomer()
        ? Limit::none()
        : Limit::perHour(1)->by($job->user->id);
});

use Illuminate\Queue\Middleware\RateLimited;

public function middleware(): array
{
    return [new RateLimited('backups')];
}

超出限流阈值的任务会根据剩余等待时间自动释放回队列。也可以通过 releaseAfter() 固定重试的间隔秒数，或使用 dontRelease() 直接结束而不再重试。

public function middleware(): array
{
    return [(new RateLimited('backups'))->releaseAfter(60)];
    // 或者不再重试的写法
    // return [(new RateLimited('backups'))->dontRelease()];
}

被释放的任务同样会计入尝试次数（attempts）。请合理配置 #[Tries] 或 retryUntil()。

如果使用 Redis，可以选用性能更好的 Illuminate\Queue\Middleware\RateLimitedWithRedis。

use Illuminate\Queue\Middleware\RateLimitedWithRedis;

public function middleware(): array
{
    return [new RateLimitedWithRedis('backups')];
}

防止重复执行（WithoutOverlapping）

Illuminate\Queue\Middleware\WithoutOverlapping 可基于任意 key 阻止同一任务被并发执行。对于同一资源一次只希望有一个任务在更新的场景非常有用。

use Illuminate\Queue\Middleware\WithoutOverlapping;

public function middleware(): array
{
    // 避免同一用户 ID 的信用评分更新任务重复执行
    return [new WithoutOverlapping($this->user->id)];
}

重复的任务会被释放回队列，可用 releaseAfter() 指定重试间隔，用 dontRelease() 让其立即被丢弃。由于底层使用原子锁机制，建议使用 expireAfter() 明确指定锁的有效期，这样即使任务意外失败或超时，锁也不会一直残留。

public function middleware(): array
{
    return [
        (new WithoutOverlapping($this->order->id))
            ->releaseAfter(60)
            ->expireAfter(180),
    ];
}

默认情况下只在同一任务类中防止重复。若希望在不同任务类之间共享锁的 key，可使用 shared() 方法。

抑制连续异常（ThrottlesExceptions）

Illuminate\Queue\Middleware\ThrottlesExceptions 用于与外部 API 等不稳定服务对接的任务：当异常达到指定次数后，会暂停后续执行一段时间。通常与基于时间的重试限制（retryUntil()）配合使用。

use DateTime;
use Illuminate\Queue\Middleware\ThrottlesExceptions;

public function middleware(): array
{
    // 连续发生 10 次异常后，暂停任务执行 5 分钟
    return [new ThrottlesExceptions(10, 5 * 60)];
}

public function retryUntil(): DateTime
{
    return now()->addMinutes(30);
}

可通过 when() 仅对特定异常进行节流，或用 deleteWhen() 在遇到某类异常时直接删除任务，实现更细粒度的控制。

use App\Exceptions\CustomerDeletedException;
use Illuminate\Http\Client\HttpClientException;
use Illuminate\Queue\Middleware\ThrottlesExceptions;

public function middleware(): array
{
    return [
        (new ThrottlesExceptions(10, 10 * 60))
            ->when(fn ($e) => $e instanceof HttpClientException)
            ->deleteWhen(CustomerDeletedException::class),
    ];
}

如果使用 Redis，可用 Illuminate\Queue\Middleware\ThrottlesExceptionsWithRedis 更高效地执行节流。

任务释放（Release 中间件）

在特定条件下不执行任务而将其放回队列，可使用 Release 中间件：

use Illuminate\Queue\Middleware\Release;

public function middleware(): array
{
    return [
        Release::when($this->order->isPending(), releaseAfter: 60),
    ];
}

Release::unless() 则在条件为 false 时释放。

return [
    Release::unless($this->order->isPaid(), releaseAfter: 60),
];

用闭包表达复杂条件：

return [
    Release::when(function (): bool {
        return ! $this->order->isPaid();
    }, releaseAfter: 60),
];

Release 也会累加尝试次数。请合理配置 #[Tries] 或 $tries。

失败任务处理

准备 failed_jobs 表

任务超过最大重试次数会写入 failed_jobs。若无此表：

php artisan make:queue-failed-table
php artisan migrate

失败时的清理

在任务类定义 failed() 方法：

use Throwable;

public function failed(?Throwable $exception): void
{
    // 例如向管理员发送 Slack 通知
}

用异常控制不重试

某些异常无需重试。在 bootstrap/app.php 的 withExceptions() 中用 dontRetry 指定：

use App\Exceptions\InvalidPodcastSourceException;
use Illuminate\Foundation\Configuration\Exceptions;

->withExceptions(function (Exceptions $exceptions): void {
    $exceptions->dontRetry([
        InvalidPodcastSourceException::class,
    ]);
})

更细粒度可用 dontRetryWhen，回调返回 true 时立即失败：

use App\Exceptions\PodcastProcessingException;
use Illuminate\Foundation\Configuration\Exceptions;

->withExceptions(function (Exceptions $exceptions): void {
    $exceptions->dontRetryWhen(function (PodcastProcessingException $e) {
        return $e->reason() === 'Subscription expired';
    });
})

校验错误或订阅过期等重试也无法成功的异常适合直接失败。

查看失败任务

php artisan queue:failed

重试失败任务

# 指定 ID 重试
php artisan queue:retry ce7bb17c-cdd8-41f0-a8ec-7b4fef4e5ece

# 全部重试
php artisan queue:retry all

删除失败任务

php artisan queue:forget ce7bb17c-cdd8-41f0-a8ec-7b4fef4e5ece
php artisan queue:flush

常用驱动

database

无需额外中间件即可上手。任务存 jobs 表，worker 轮询处理。

优点：配置简单，可复用 RDBMS
缺点：数据库压力大，不适合大量任务

QUEUE_CONNECTION=database

redis

生产环境最常用，内存速度快，可支撑大量任务。

优点：速度快、可扩展
缺点：需要 Redis 服务

QUEUE_CONNECTION=redis
REDIS_HOST=127.0.0.1
REDIS_PORT=6379

在生产环境使用 Redis 队列时，可考虑引入 Laravel Horizon，通过仪表盘实时监控队列。

使用 Supervisor 在生产环境运行

在 Linux 上通常使用 Supervisor 保证 queue:work 崩溃后自动重启：

# /etc/supervisor/conf.d/laravel-worker.conf
[program:laravel-worker]
process_name=%(program_name)s_%(process_num)02d
command=php /var/www/your-app/artisan queue:work --sleep=3 --tries=3 --max-time=3600
autostart=true
autorestart=true
stopasgroup=true
killasgroup=true
user=www-data
numprocs=2
redirect_stderr=true
stdout_logfile=/var/www/your-app/storage/logs/worker.log
stopwaitsecs=3600

numprocs=2 表示并行启动 2 个 worker。之后：

sudo supervisorctl reread
sudo supervisorctl update
sudo supervisorctl start laravel-worker:*

实战示例：用队列处理邮件发送

生成任务类

php artisan make:job SendOrderConfirmation

实现任务

<?php

namespace App\Jobs;

use App\Models\Order;
use App\Mail\OrderConfirmed;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Queue\Queueable;
use Illuminate\Queue\Attributes\Tries;
use Illuminate\Queue\Attributes\Timeout;
use Illuminate\Support\Facades\Mail;

#[Tries(3)]
#[Timeout(30)]
class SendOrderConfirmation implements ShouldQueue
{
    use Queueable;

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

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

控制器中派发

use App\Jobs\SendOrderConfirmation;

public function store(Request $request): RedirectResponse
{
    $order = Order::create($request->validated());

    SendOrderConfirmation::dispatch($order);

    return redirect()->route('orders.show', $order)
        ->with('success', '订单已受理。');
}

启动 worker

php artisan queue:work --tries=3 --timeout=30

小结

何时应使用队列

发送邮件 / SMS
图片、视频的缩放或转码
请求外部 API
生成报表或导出 CSV
发送 Webhook

开发小技巧

在 .env 中设置 QUEUE_CONNECTION=sync，任务将立即执行、不经队列，无需启动 worker 就能调试。

QUEUE_CONNECTION=sync

常用命令

# 启动 worker
php artisan queue:work

# 部署后重启 worker
php artisan queue:restart

# 失败任务列表
php artisan queue:failed

# 全部重试
php artisan queue:retry all

# 全部删除
php artisan queue:flush

​什么是队列

​队列配置

​config/queue.php

​.env 配置

​准备数据库驱动

​准备 Redis 驱动

​SQS Overflow Storage

​创建任务类

​make:job 命令

​任务类结构

​派发任务

​dispatch()

​延迟派发

​dispatchAfterResponse()

​指定队列

​Queue Routing

​队列转发（Queue::forward()）

​同步执行（测试 / 开发）

​批量派发

​任务链

​任务批处理

​处理任务

​queue:work

​worker 常用选项

​在任务类上写重试配置

​使用任务中间件控制执行

​速率限制（RateLimited）

​防止重复执行（WithoutOverlapping）

​抑制连续异常（ThrottlesExceptions）

​任务释放（Release 中间件）

​失败任务处理

​准备 failed_jobs 表

​失败时的清理

​用异常控制不重试

​查看失败任务

​重试失败任务

​删除失败任务

​常用驱动

​database

​redis

​使用 Supervisor 在生产环境运行

​实战示例：用队列处理邮件发送

​小结

什么是队列

队列配置

config/queue.php

.env 配置

准备数据库驱动

准备 Redis 驱动

SQS Overflow Storage

创建任务类

make:job 命令

任务类结构

派发任务

dispatch()

延迟派发

dispatchAfterResponse()

指定队列

Queue Routing

队列转发（`Queue::forward()`）

同步执行（测试 / 开发）

批量派发

任务链

任务批处理

处理任务

queue:work

worker 常用选项

在任务类上写重试配置

使用任务中间件控制执行

速率限制（RateLimited）

防止重复执行（WithoutOverlapping）

抑制连续异常（ThrottlesExceptions）

任务释放（Release 中间件）

失败任务处理

准备 failed_jobs 表

失败时的清理

用异常控制不重试

查看失败任务

重试失败任务

删除失败任务

常用驱动

database

redis

使用 Supervisor 在生产环境运行

实战示例：用队列处理邮件发送

小结