Laravel: Laravel队列与异步任务

Q: Batch 任务失败了一个怎么办？

默认 Batch 遇到第一个失败就取消剩余任务。设置 allowFailures() 允许部分失败继续执行。用 Bus::findBatch() 查看失败详情，手动重试失败的任务。

Q: 如何调试 Job？

开发时用 QUEUE_CONNECTION=sync 同步执行（错误直接显示）；生产用 php artisan queue:failed 查看失败 Job 的错误信息；也可以在 Job 的 failed() 方法中记录详细日志。

最后更新：2026-08-26

队列是 Laravel 的"后台团队"——耗时任务丢给队列异步处理，用户不用等待，请求秒回。

1. 你将学到

队列驱动配置：sync/database/redis/sqs 对比
Job 创建与分发：dispatch() / Queue::push()
失败重试机制：tries/backoff/failed_jobs 表
队列优先级、延迟与批量任务：Batch/Bused
Supervisor 守护 queue:work 进程

2. 一个用户的故事

(1) 痛点：导出报表让整个系统卡住

Bob 在 ShopMetrics 后台点击"导出月度报表"——10 万条订单数据生成 CSV 需要 30 秒，期间页面一直转圈，Alice 同时访问仪表盘也变慢了。更糟的是，Bob 同时触发了 5 个报表导出，PHP 进程池被耗尽，整个站点 502。

(2) 队列的解法

队列把耗时任务丢到后台——用户点击导出后立即收到"报表生成中"提示，Job 在后台慢慢跑，完成后邮件通知下载链接。

PHP

// Sync — blocks for 30 seconds
$csv = ReportService::generateMonthlyReport($tenant);

// Async — returns instantly, Job runs in background
GenerateReportJob::dispatch($tenant, 'monthly');
// User sees: "Report is being generated. We'll email you when it's ready."

(3) 收益

Bob 用队列后，报表导出请求 100ms 内返回，后台 Job 慢慢跑，Alice 的仪表盘再也不会卡。

3. 队列驱动配置

(1) 驱动对比

驱动	持久化	性能	适合	成本
`sync`	❌ 即时执行	最快	开发/测试	免费
`database`	✅ DB 表	慢	小项目	免费
`redis`	✅ 内存	快	生产环境	Redis
`sqs`	✅ AWS	高	大规模	按量
`beanstalkd`	✅ 专用	快	中等	免费

(2) 配置

BASH

# .env
QUEUE_CONNECTION=redis

# Redis connection
REDIS_HOST=127.0.0.1
REDIS_PASSWORD=null
REDIS_PORT=6379

PHP

// config/queue.php
'connections' => [
    'database' => [
        'driver' => 'database',
        'table' => 'jobs',
        'queue' => 'default',
        'retry_after' => 90,
    ],
    'redis' => [
        'driver' => 'redis',
        'connection' => 'default',
        'queue' => '{default}',
        'retry_after' => 90,
        'block_for' => null,
    ],
],

(3) 创建队列表

BASH

# For database driver
php artisan queue:table
php artisan queue:failed-table
php artisan migrate

# Creates: jobs table + failed_jobs table

▶ 示例：ShopMetrics 队列配置

BASH

# .env — Development
QUEUE_CONNECTION=database

# .env — Production
QUEUE_CONNECTION=redis
REDIS_HOST=redis.shopmetrics.internal
REDIS_PORT=6379

# Create queue tables (database driver)
php artisan queue:table
php artisan queue:failed-table
php artisan queue:batches-table
php artisan migrate

输出：

TEXT 📖 仅展示

# 命令执行成功

4. Job 创建与分发

(1) 创建 Job

BASH

php artisan make:job GenerateReportJob
# Creates: app/Jobs/GenerateReportJob.php

(2) 定义 Job

PHP

// app/Jobs/GenerateReportJob.php
class GenerateReportJob implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;

    public int $tries = 3;
    public int $backoff = 60;
    public bool $deleteWhenMissingModels = true;

    public function __construct(
        public Tenant $tenant,
        public string $reportType,
        public string $format = 'csv',
    ) {}

    public function handle(
        ReportService $reportService,
        Mailer $mailer,
    ): void {
        $path = $reportService->generate(
            $this->tenant,
            $this->reportType,
            $this->format,
        );

        $url = Storage::disk('s3')->temporaryUrl($path, now()->addDays(7));

        $this->tenant->users->each(function ($user) use ($url) {
            $mailer->to($user)->send(new ReportReadyNotification($url, $this->reportType));
        });
    }

    public function failed(\Throwable $exception): void
    {
        Log::error('Report generation failed', [
            'tenant_id' => $this->tenant->id,
            'report_type' => $this->reportType,
            'error' => $exception->getMessage(),
        ]);
    }
}

(3) 分发 Job

PHP

// Basic dispatch
GenerateReportJob::dispatch($tenant, 'monthly');

// Delayed dispatch
GenerateReportJob::dispatch($tenant, 'monthly')
    ->delay(now()->addMinutes(5));

// Dispatch to specific queue
GenerateReportJob::dispatch($tenant, 'monthly')
    ->onQueue('reports');

// Dispatch if condition met
GenerateReportJob::dispatchIf($tenant->subscription?->isActive(), $tenant, 'monthly');

// Dispatch after response sent to user
GenerateReportJob::dispatchAfterResponse($tenant, 'monthly');

分发方式	说明	适用场景
`dispatch()`	推入队列	一般异步任务
`dispatchSync()`	同步执行	必须立即完成
`dispatchAfterResponse()`	响应后执行	轻量任务
`delay()`	延迟执行	定时任务
`onQueue()`	指定队列	分优先级

▶ 示例：ShopMetrics 订单处理 Job

PHP

// app/Jobs/ProcessOrderJob.php
class ProcessOrderJob implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;

    public int $tries = 3;
    public int $backoff = [30, 60, 120];

    public function __construct(public Order $order) {}

    public function handle(
        OrderService $orderService,
        PaymentGateway $payment,
    ): void {
        // Charge payment
        $payment->charge($this->order);

        // Update inventory
        $orderService->deductInventory($this->order);

        // Broadcast event
        event(new OrderPlaced($this->order));

        // Send confirmation email
        $this->order->user->notify(new OrderConfirmationNotification($this->order));
    }

    public function failed(\Throwable $exception): void
    {
        $this->order->update(['status' => 'failed']);
        $this->order->user->notify(new OrderFailedNotification($this->order));
    }
}

输出：

TEXT 📖 仅展示

// 执行成功

5. 失败重试机制

(1) 重试配置

PHP

// In Job class
public int $tries = 3;           // Max retry attempts
public int $backoff = 60;        // Seconds between retries
public int $timeout = 120;       // Max seconds per attempt

// Or exponential backoff
public array $backoff = [30, 60, 120]; // 30s, 60s, 120s

// Or dynamic backoff
public function backoff(): int
{
    return 30 * $this->attempts();
}

(2) 失败处理

BASH

# View failed jobs
php artisan queue:failed

# Retry a specific failed job
php artisan queue:retry 5

# Retry all failed jobs
php artisan queue:retry all

# Delete a failed job
php artisan queue:forget 5

# Clear all failed jobs
php artisan queue:flush

(3) 队列任务生命周期

flowchart LR
    A[Job Dispatched] --> B[Queue]
    B --> C[Worker Picks Up]
    C --> D{Success?}
    D -->|Yes| E[Job Completed]
    D -->|No| F{attempts < tries?}
    F -->|Yes| G[Backoff + Retry]
    G --> B
    F -->|No| H[failed_jobs Table]
    H --> I[Manual Retry / Flush]

▶ 示例：ShopMetrics 失败重试配置

PHP

// app/Jobs/SendWebhookJob.php
class SendWebhookJob implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;

    public int $tries = 5;
    public array $backoff = [10, 30, 60, 120, 300];
    public int $timeout = 30;

    public function __construct(
        public Shop $shop,
        public array $payload,
    ) {}

    public function handle(): void
    {
        $response = Http::timeout($this->timeout)
            ->post($this->shop->webhook_url, $this->payload);

        if (!$response->successful()) {
            $this->release($this->backoff[$this->attempts() - 1] ?? 60);
        }
    }

    public function failed(\Throwable $exception): void
    {
        $this->shop->tenant->users->each(function ($user) {
            $user->notify(new WebhookFailedNotification($this->shop));
        });
    }
}

输出：

TEXT 📖 仅展示

// 执行成功

6. 批量任务

(1) 创建 Batch

BASH

php artisan queue:batches-table
php artisan migrate

PHP

// app/Jobs/ProcessTenantAnalyticsJob.php
class ProcessTenantAnalyticsJob implements ShouldQueue
{
    use Batchable;

    public function __construct(public Tenant $tenant) {}

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

        AnalyticsService::computeForTenant($this->tenant);
    }
}

(2) 分发 Batch

PHP

use Illuminate\Bus\Batch;
use Illuminate\Support\Facades\Bus;

$batch = Bus::batch(
    Tenant::active()->get()->map(
        fn ($tenant) => new ProcessTenantAnalyticsJob($tenant)
    )
)->then(function (Batch $batch) {
    // All jobs completed successfully
    Log::info("Batch {$batch->id} completed: {$batch->totalJobs} tenants processed.");
})->catch(function (Batch $batch, \Throwable $e) {
    // First job failure
    Log::error("Batch {$batch->id} failed: {$e->getMessage()}");
})->finally(function (Batch $batch) {
    // Always runs (success or failure)
    Cache::forget('analytics:computing');
})->name('Process Monthly Analytics')
  ->onQueue('analytics')
  ->dispatch();

(3) Batch 管理

PHP

// Check batch progress
$batch = Bus::findBatch($batchId);
$batch->progress();     // 0-100
$batch->processedJobs();
$batch->totalJobs();
$batch->failedJobs();
$batch->finished();

// Cancel batch
$batch->cancel();

方法	说明
`then()`	全部成功回调
`catch()`	首次失败回调
`finally()`	完成后回调（无论成败）
`progress()`	进度百分比
`cancel()`	取消剩余任务

▶ 示例：ShopMetrics 月度分析批量任务

PHP

// app/Console/Commands/ProcessMonthlyAnalytics.php
class ProcessMonthlyAnalytics extends Command
{
    protected $signature = 'analytics:process-monthly';

    public function handle(): int
    {
        $tenants = Tenant::active()->get();
        $this->info("Processing analytics for {$tenants->count()} tenants...");

        Bus::batch(
            $tenants->map(fn ($t) => new ProcessTenantAnalyticsJob($t))
        )->then(function (Batch $batch) {
            $this->info("All {$batch->totalJobs} tenants processed.");
        })->catch(function (Batch $batch, \Throwable $e) {
            $this->error("Batch failed: {$e->getMessage()}");
        })->name('Monthly Analytics')
          ->onQueue('analytics')
          ->allowFailures()
          ->dispatch();

        return self::SUCCESS;
    }
}

输出：

TEXT 📖 仅展示

// 执行成功

7. Supervisor 守护

(1) 安装 Supervisor

BASH

# Ubuntu/Debian
sudo apt-get install supervisor

# Create config
sudo nano /etc/supervisor/conf.d/shopmetrics-worker.conf

(2) Supervisor 配置

INI

[program:shopmetrics-worker-default]
process_name=%(program_name)s_%(process_num)02d
command=php /var/www/shopmetrics/artisan queue:work redis --queue=default --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/log/shopmetrics/worker-default.log
stopwaitsecs=3600

[program:shopmetrics-worker-reports]
process_name=%(program_name)s_%(process_num)02d
command=php /var/www/shopmetrics/artisan queue:work redis --queue=reports --sleep=3 --tries=3 --max-time=3600
autostart=true
autorestart=true
numprocs=1
user=www-data
redirect_stderr=true
stdout_logfile=/var/log/shopmetrics/worker-reports.log

(3) Supervisor 命令

BASH

# Read new config
sudo supervisorctl reread
sudo supervisorctl update

# Start/stop/restart workers
sudo supervisorctl start shopmetrics-worker-default:*
sudo supervisorctl stop shopmetrics-worker-default:*
sudo supervisorctl restart shopmetrics-worker-default:*

# Check status
sudo supervisorctl status

进程数	队列	CPU	内存	说明
2	default	低	中	通用任务
1	reports	高	高	报表生成
1	analytics	高	高	数据分析
1	notifications	低	低	邮件/推送

▶ 示例：ShopMetrics 多队列 Worker 启动

BASH

# Development — single worker, all queues
php artisan queue:work --queue=default,reports,notifications

# Production — separate workers per queue priority
# Priority: high > default > low
php artisan queue:work redis --queue=high,default
php artisan queue:work redis --queue=reports,low

# Process jobs with time limit (auto-restart for memory leaks)
php artisan queue:work --max-time=3600 --max-jobs=1000

# Monitor queue
php artisan queue:monitor redis:default,redis:reports

输出：

TEXT 📖 仅展示

# 命令执行成功

8. 综合示例：ShopMetrics 异步报表系统

PHP

// ============================================
// Comprehensive: ShopMetrics Async Report System
// Covers: jobs, batches, retries, supervisor, notifications
// ============================================

// app/Jobs/GenerateReportJob.php
class GenerateReportJob implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;

    public int $tries = 3;
    public array $backoff = [60, 180, 600];
    public int $timeout = 600;
    public bool $deleteWhenMissingModels = true;

    public function __construct(
        public Tenant $tenant,
        public string $reportType,
        public string $format = 'csv',
        public ?int $userId = null,
    ) {
        $this->onQueue('reports');
    }

    public function handle(ReportService $reportService): void
    {
        $path = $reportService->generate(
            $this->tenant,
            $this->reportType,
            $this->format,
        );

        $downloadUrl = Storage::disk('s3')->temporaryUrl($path, now()->addDays(7));

        $user = $this->userId ? User::find($this->userId) : $this->tenant->users->first();
        $user?->notify(new ReportReadyNotification(
            downloadUrl: $downloadUrl,
            reportType: $this->reportType,
            expiresAt: now()->addDays(7),
        ));
    }

    public function failed(\Throwable $exception): void
    {
        $user = $this->userId ? User::find($this->userId) : $this->tenant->users->first();
        $user?->notify(new ReportFailedNotification(
            reportType: $this->reportType,
            error: $exception->getMessage(),
        ));
    }
}

// Usage in controller
class ReportController extends Controller
{
    public function generate(Request $request): JsonResponse
    {
        $validated = $request->validate([
            'report_type' => 'required|in:monthly,weekly,custom',
            'format' => 'sometimes|in:csv,xlsx,pdf',
            'date_from' => 'sometimes|date',
            'date_to' => 'sometimes|date|after:date_from',
        ]);

        GenerateReportJob::dispatch(
            tenant(),
            $validated['report_type'],
            $validated['format'] ?? 'csv',
            auth()->id(),
        );

        return response()->json([
            'message' => 'Report generation started. You will receive an email when ready.',
            'estimated_time' => '5-15 minutes',
        ], 202);
    }
}

❓ 常见问题

Q queue:work 和 queue:listen 有什么区别？

A queue:work 常驻内存不重启框架（快但代码变更需手动重启）；queue:listen 每个任务重启框架（慢但自动加载新代码）。生产用 queue:work + Supervisor，开发用 queue:listen 或 queue:work --once。

Q Job 里能用 Eloquent 模型吗？

A 可以，SerializesModels trait 自动序列化模型 ID，反序列化时从数据库重新获取。好处是 Job 数据量小；风险是如果模型被删除，Job 会失败（可用 deleteWhenMissingModels 忽略）。

Q 如何防止队列堆积？

A 增加 Worker 数量（numprocs）、使用更高性能的驱动（Redis）、设置 Job timeout 避免卡住、监控队列长度告警。

Q Batch 任务失败了一个怎么办？

A 默认 Batch 遇到第一个失败就取消剩余任务。设置 allowFailures() 允许部分失败继续执行。用 Bus::findBatch() 查看失败详情，手动重试失败的任务。

Q 如何调试 Job？

A 开发时用 QUEUE_CONNECTION=sync 同步执行（错误直接显示）；生产用 php artisan queue:failed 查看失败 Job 的错误信息；也可以在 Job 的 failed() 方法中记录详细日志。

Q Supervisor 和 systemd 有什么区别？

A Supervisor 是 Python 进程管理器，配置简单，Laravel 官方推荐；systemd 是 Linux 原生服务管理器，性能更好但配置更复杂。两者都能守护 queue:work 进程。

📖 小节

队列把耗时任务异步化，用户请求秒回
Redis 是生产环境最佳队列驱动，性能高且持久化
Job 用 tries/backoff 配置重试策略，exponential backoff 避免雪崩
Batch 处理大量同类任务，支持进度追踪和部分失败
Supervisor 守护 Worker 进程，崩溃自动重启
多队列分优先级：high/default/reports/low

📝 作业

基础题（⭐）：创建 GenerateReportJob，在控制器中 dispatch 异步生成 CSV 报表，使用 database 驱动测试，运行 queue:work 确认 Job 执行。
进阶题（⭐⭐）：配置 Redis 队列驱动，实现 3 次重试 + exponential backoff（30s/60s/120s），在 Job 失败时发送通知给用户，测试手动重试（queue:retry）。
挑战题（⭐⭐⭐）：使用 Bus::batch() 实现多租户月度分析批量任务——遍历所有活跃租户，每个租户一个 Job，监控进度百分比，完成后清理缓存，配置 Supervisor 守护 Worker。

Web-Tutorial 技术团队

由多位开发者共同维护的编程教程平台。每篇教程由对应领域的开发者编写和审核，确保内容准确可靠。如发现任何问题，欢迎向我们反馈。

关于我们

Laravel: Laravel队列与异步任务

1. 你将学到

2. 一个用户的故事

(1) 痛点：导出报表让整个系统卡住

(2) 队列的解法

(3) 收益

3. 队列驱动配置

(1) 驱动对比

(2) 配置

(3) 创建队列表

▶ 示例：ShopMetrics 队列配置

4. Job 创建与分发

(1) 创建 Job

(2) 定义 Job

(3) 分发 Job

▶ 示例：ShopMetrics 订单处理 Job

5. 失败重试机制

(1) 重试配置

(2) 失败处理

(3) 队列任务生命周期

▶ 示例：ShopMetrics 失败重试配置

6. 批量任务

(1) 创建 Batch

(2) 分发 Batch

(3) Batch 管理

▶ 示例：ShopMetrics 月度分析批量任务

7. Supervisor 守护

(1) 安装 Supervisor

(2) Supervisor 配置

(3) Supervisor 命令

▶ 示例：ShopMetrics 多队列 Worker 启动

8. 综合示例：ShopMetrics 异步报表系统

❓ 常见问题

📖 小节

📝 作业

🙏 帮我们做得更好