队列与任务 - 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 => ['podcasts', 'redis'],
    ProcessVideo::class => 'videos',
]);

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

同步执行（测试 / 开发）

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() 不同，没有进度追踪与完成回调，适合大量独立任务的简单批量发送。

处理任务

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;

    // ...
}

任务释放（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:work

​worker 常用选项

​在任务类上写重试配置

​任务释放（Release 中间件）

​失败任务处理

​准备 failed_jobs 表

​失败时的清理

​用异常控制不重试

​查看失败任务

​重试失败任务

​删除失败任务

​常用驱动

​database

​redis

​使用 Supervisor 在生产环境运行

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

​小结

什么是队列

队列配置

config/queue.php

.env 配置

准备数据库驱动

准备 Redis 驱动

SQS Overflow Storage

创建任务类

make:job 命令

任务类结构

派发任务

dispatch()

延迟派发

dispatchAfterResponse()

指定队列

Queue Routing

同步执行（测试 / 开发）

批量派发

处理任务

queue:work

worker 常用选项

在任务类上写重试配置

任务释放（Release 中间件）

失败任务处理

准备 failed_jobs 表

失败时的清理

用异常控制不重试

查看失败任务

重试失败任务

删除失败任务

常用驱动

database

redis

使用 Supervisor 在生产环境运行

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

小结