Queue 與 Job - Knowledge of Laravel

什麼是 Queue

在 Web 應用程式中，會有寄信、縮圖、對外部 API 查詢等需要數秒才能完成的處理。若在 HTTP 請求中同步進行這些處理，使用者必須一直等到回應返回。使用 Laravel 的 Queue，能將這些重負載處理在背景以非同步方式執行。請求會立刻回應，實際處理由 worker process 另外執行。

Queue 支援資料庫、Redis、Amazon SQS 等多種後端。開發環境使用 sync driver 時，不透過 queue 即可立即執行 job。

Queue 的設定

config/queue.php

Queue 的設定集中於 config/queue.php。以 QUEUE_CONNECTION 環境變數切換所使用的 driver。

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

.env 設定

# 選擇 driver
QUEUE_CONNECTION=database

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

資料庫 driver 的準備

使用 database driver 時，需要儲存 job 的資料表。 Laravel 11 以後的新專案已預設包含 migration，若沒有，可用以下指令建立：

php artisan make:queue-table
php artisan migrate

Redis driver 的準備

使用 redis driver 時，於 config/database.php 加入 Redis 連線設定，並以 Composer 安裝 driver：

composer require predis/predis

SQS Overflow Storage

Amazon SQS 的訊息 payload 有大小上限。若要處理較大的 payload，可加入將超過的部分存到 cache store、只將 pointer 傳給 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 的 payload 存到指定的 cache store。
將 always 設為 true，則無論大小都會把所有 SQS payload 存到 cache store。
delete_after_processing 會在 job 成功後刪除已儲存的 payload（預設 true）。
將 flush_on_clear 設為 true，執行 queue:clear 時會 flush overflow 用的 store。因會避免清掉一般 cache，建議搭配專用 store 使用。

建立 Job 類別

make:job 指令

以 make:job Artisan 指令產生 job 類別的樣板：

php artisan make:job SendWelcomeEmail

會產生 app/Jobs/SendWelcomeEmail.php。

Job 類別的結構

<?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;

    /**
     * 建立 job 的實例
     */
    public function __construct(
        public User $user,
    ) {}

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

實作 ShouldQueue 介面，是在告訴 Laravel 這個 job 要以 queue 非同步處理。 Queueable trait 提供 job 佇列操作所需的方法。

在建構子傳入 Eloquent model 時，Laravel 會自動只序列化 ID。執行時會再從資料庫重新取得最新資料，因此 queue 的 payload 較輕。

Job 的分派

dispatch()

從 controller 或 service 將 job 送到 queue，使用 dispatch()：

use App\Jobs\SendWelcomeEmail;

// 於路由或 controller 中
public function register(Request $request): RedirectResponse
{
    $user = User::create($request->validated());

    // 將 job 放入 queue
    SendWelcomeEmail::dispatch($user);

    return redirect('/dashboard');
}

延遲 dispatch

delay() 方法可讓 job 的執行延後指定時間。

// 5 分鐘後執行
SendWelcomeEmail::dispatch($user)->delay(now()->addMinutes(5));

dispatchAfterResponse()

使用 dispatchAfterResponse() 會在回應 HTTP 給使用者之後立即執行 job。在 sync driver 下也能運作，適合不需要專用 worker 的輕量用途。

SendWelcomeEmail::dispatchAfterResponse($user);

分派到特定 queue

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

Queue Routing

若要將特定 job 類別預設導向指定 connection／queue，可在 ServiceProvider 的 boot() 中使用 Queue::route()。無需為每個 job 類別加上 onQueue() / onConnection()，可集中管理。

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、父類別。所有實作、使用、繼承這些的 job 都會自動套用。若要一次路由多個 job，傳入陣列：

Queue::route([
    ProcessPodcast::class => ['redis', 'podcasts'], // connection 與 queue
    ProcessVideo::class => 'videos',                // 只有 queue（使用預設 connection）
]);

Queue Routing 可被 job 端的 onQueue() / onConnection() 覆寫。

Queue 轉發（`Queue::forward()`）

使用 Queue::forward() 可以將 job 從某個 queue 轉發到其他 queue／connection。當你想切換 queue 基礎架構而不修改個別 job 或呼叫端程式碼時非常方便。

use Illuminate\Support\Facades\Queue;

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

若要一次轉發多個 queue，可傳入陣列。

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

若 job 端已明確指定 connection，則 job 的指定會優先於轉發設定。

同步執行（用於測試／開發）

使用 dispatchSync() 會跳過 queue 立即執行。

SendWelcomeEmail::dispatchSync($user);

大量 dispatch

當要一次分派多個獨立的 job 時，可使用 Bus facade 的 bulk() 方法。適合不需要如 batch 處理般的追蹤或 callback 的情境。 Bus::bulk() 會依所設定的 queue connection 與 queue 名稱將 job 分組，並將每組整批推入 queue，因此效率較高。

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

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

Bus::bulk() 會將 job 以批次方式送入 queue。與 batch 處理（Bus::batch()）不同，不提供進度追蹤或完成 callback。適合需要簡潔地批次送出大量獨立 job 的情境。

Job Chain

將 job 串成 chain 後，可依序執行多個 job。當 chain 內的某個 job 失敗時，後續的 job 就不會執行。

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();

也可以加入在整條 chain 完成時、或 chain 內的 job 失敗時執行的 callback。

Bus::chain([
    new ProcessPodcast($podcast),
    new ReleasePodcast($podcast),
])->catch(function (Throwable $e) {
    // chain 內的 job 失敗時的處理
})->dispatch();

Job Batch

使用 batch 可一次分派多個 job，並追蹤整體進度。首先建立 job_batches 資料表用的 migration。

php artisan make:batches-table
php artisan migrate

在 job 類別中使用 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;
        }

        // 匯入聯絡人的 chunk...
    }
}

以 Bus::batch() 分派 batch，並可註冊完成、失敗、結束時的 callback。

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) {
    // 所有 job 皆正常完成時
})->catch(function (Batch $batch, Throwable $e) {
    // 有 job 失敗時
})->finally(function (Batch $batch) {
    // batch 執行結束時
})->dispatch();

可以使用 batch ID 確認 batch 的狀態。

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

$batch->totalJobs;      // job 總數
$batch->pendingJobs;    // 等待中的 job 數
$batch->failedJobs;     // 失敗的 job 數
$batch->progress();     // 進度（0〜100）

Job 的處理

queue:work 指令

啟動 queue worker 來處理 job。

php artisan queue:work

也可指定 driver 或 queue。

# 只處理 Redis 的 emails queue
php artisan queue:work redis --queue=emails

# 使用 database driver
php artisan queue:work database

queue:work 啟動後會持續運行。變更程式碼時請以 queue:restart 重啟 worker。正式環境一般會以 Supervisor 等 process manager 管理。

Queue Worker 的監控選項

可組合常用選項精細控制 worker：

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

選項	說明
`--tries=N`	Job 的最大嘗試次數。超過會作為失敗 job 記錄
`--timeout=N`	單一 job 的最大執行秒數。超過會強制結束 worker
`--sleep=N`	queue 為空時到下次輪詢的等待秒數（預設：3）
`--max-jobs=N`	處理 N 個後結束 worker
`--max-time=N`	經過 N 秒後結束 worker
`--queue=A,B`	依優先順序處理 queue（A 優先）

在 Job 類別中設定重試

比起命令列選項，將設定寫在 job 類別本身有時更易於管理。

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;

    // ...
}

將崩潰計為例外

預設情況下，因記憶體不足等原因導致 worker 行程崩潰或被強制終止而結束的嘗試，不會計入 job 的最大例外數（MaxExceptions）。若希望這些嘗試也計為例外，請在 job 類別中加上 CountCrashesAsExceptions 屬性。

use Illuminate\Queue\Attributes\CountCrashesAsExceptions;
use Illuminate\Queue\Attributes\MaxExceptions;
use Illuminate\Queue\Attributes\Tries;

#[Tries(25)]
#[MaxExceptions(3)]
#[CountCrashesAsExceptions]
class ProcessPodcast implements ShouldQueue
{
    // ...
}

加上此屬性後，worker 在處理 job 期間會將標記儲存至應用程式的快取。下次嘗試該 job 時若標記仍存在，前一次嘗試便會計為例外。

以 Job Middleware 控制執行

Job middleware 可將速率限制、防止重疊執行等橫向邏輯從 handle() 方法中抽離，改由 middleware() 方法宣告式地指定。由於邏輯不寫在 job 本身，多個 job 也能輕鬆重複使用相同的控制。

/**
 * 回傳 job 會經過的 middleware
 */
public function middleware(): array
{
    return [new RateLimited('backups')];
}

自訂的 job middleware 可用 make:job-middleware Artisan 指令產生。Job middleware 也能套用到queued event listener、mailable、通知上。

速率限制（RateLimited）

使用 RateLimiter facade 的 for 方法定義速率限制，再將 Illuminate\Queue\Middleware\RateLimited middleware 套用到 job 上。

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')];
}

超過限制的 job 會依照速率限制的剩餘時間，自動被 release 回 queue。可用 releaseAfter() 固定重試前的秒數，或用 dontRelease() 不再重試而直接結束。

public function middleware(): array
{
    return [(new RateLimited('backups'))->releaseAfter(60)];
    // 或者不進行重試時
    // return [(new RateLimited('backups'))->dontRelease()];
}

被 release 的 job 嘗試次數（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 為基準，避免同一個 job 同時有多個實例在執行。適用於某項資源一次只想被一個 job 更新的情境。

use Illuminate\Queue\Middleware\WithoutOverlapping;

public function middleware(): array
{
    // 避免同一使用者的信用評分更新 job 重疊執行
    return [new WithoutOverlapping($this->user->id)];
}

重疊的 job 會被 release 回 queue，可以用 releaseAfter() 指定重試間隔，或用 dontRelease() 立即刪除。由於此鎖是利用原子鎖功能實作的，建議搭配 expireAfter() 明確設定鎖的有效期，這樣即使 job 意外失敗或超時，鎖也不會一直殘留。

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

預設只會在同一個 job 類別內防止重疊執行。若想在不同 job 類別之間共用鎖 key，請使用 shared() 方法。

抑制連續發生的例外（ThrottlesExceptions）

Illuminate\Queue\Middleware\ThrottlesExceptions 用於會呼叫外部 API 等不穩定服務的 job：當累計發生指定次數例外後，暫時停止後續執行。一般會搭配以時間為基準的嘗試限制（retryUntil()）一起使用。

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

public function middleware(): array
{
    // 連續發生 10 次例外時，暫停 job 執行 5 分鐘
    return [new ThrottlesExceptions(10, 5 * 60)];
}

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

可用 when() 僅將特定例外納入節流對象，或用 deleteWhen() 在遇到特定例外時直接刪除 job，做更細緻的控制。

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 可以更有效率地進行節流。

Job 的 Release（Release middleware）

當在特定條件下想不執行 job 而放回 queue 時，使用 Release middleware 可簡潔實作。

use Illuminate\Queue\Middleware\Release;

/**
 * 回傳 job 會經過的 middleware
 */
public function middleware(): array
{
    return [
        // 當 $condition 為 true 時，60 秒後 release
        Release::when($this->order->isPending(), releaseAfter: 60),
    ];
}

Release::unless() 會在條件為 false 時 release。

return [
    // 當訂單尚未付款時，60 秒後 release
    Release::unless($this->order->isPaid(), releaseAfter: 60),
];

使用 closure 可寫更複雜的條件：

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

即使 release job，嘗試次數仍會累加。請適當設定 #[Tries] 或 $tries 屬性。

失敗 Job 的處理

準備 failed_jobs 資料表

當 job 超過最大嘗試次數，會記錄到 failed_jobs 資料表。若沒有此表，可用以下指令建立：

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

失敗時的收尾

在 job 定義 failed() 方法，可撰寫失敗時的收尾處理。

use Throwable;

public function failed(?Throwable $exception): void
{
    // 向管理員送出 Slack 通知等
    // Notification::route('slack', config('app.slack_webhook'))
    //     ->notify(new JobFailedNotification($this, $exception));
}

依例外停止重試

有些例外類型希望不再重試而直接視為失敗。在 bootstrap/app.php 的 withExceptions() 中以 dontRetry 指定要對應的例外類別。

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

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

若需更細緻的控制，可將 closure 傳給 dontRetryWhen。當 closure 回傳 true，job 會立即標記為失敗、不再重試。

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

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

對於像驗證錯誤或付款失敗（訂閱到期等）這類重試也不會改變結果的例外，用此方式立即失敗會更有效率。

檢視失敗 job 清單

php artisan queue:failed

重試失敗 job

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

# 重試所有失敗 job
php artisan queue:retry all

刪除失敗 job

# 刪除特定 job
php artisan queue:forget ce7bb17c-cdd8-41f0-a8ec-7b4fef4e5ece

# 刪除所有失敗 job
php artisan queue:flush

常用的 queue driver

database driver

無需額外的中介軟體即可使用的簡單 driver。會將 job 存到 jobs 資料表，由 worker 輪詢並處理。

優點：安裝簡單，可直接沿用既有 RDBMS
缺點：對資料庫負擔大，不適合大量 job

QUEUE_CONNECTION=database

redis driver

在正式環境中最常使用的高速 driver。以記憶體運作，吞吐量高於資料庫，可處理大量 job。

優點：高速、可擴展
缺點：需要準備 Redis 伺服器

QUEUE_CONNECTION=redis
REDIS_HOST=127.0.0.1
REDIS_PORT=6379

若在正式環境運行 Redis queue，可考慮導入 Laravel Horizon。可在漂亮的儀表板即時監控 job 狀況。

以 Supervisor 進行正式運行

在正式環境中，需要一種當 queue:work process 因某原因停止時能自動重啟的機制。 Linux 環境中一般使用 Supervisor。

# /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 process。設定後重新載入 Supervisor：

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

實務範例：以 queue 處理寄信

建立 job 類別

php artisan make:job SendOrderConfirmation

實作 job 的處理

<?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));
    }
}

從 controller 分派

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

總結

適合使用 queue 的時機

寄送 email／SMS
圖片、影片的縮圖或格式轉換
向外部 API 送出請求
產生報表或 CSV 匯出
Webhook 的送出

開發時的訣竅

將 .env 設為 QUEUE_CONNECTION=sync，job 就會不經過 queue 立即執行。不用啟動 worker 就能確認運作，開發中很方便。

QUEUE_CONNECTION=sync

常用指令總覽

# 啟動 worker
php artisan queue:work

# 重啟 worker（部署後）
php artisan queue:restart

# 失敗 job 清單
php artisan queue:failed

# 重試所有失敗 job
php artisan queue:retry all

# 刪除所有失敗 job
php artisan queue:flush

​什麼是 Queue

​Queue 的設定

​config/queue.php

​.env 設定

​資料庫 driver 的準備

​Redis driver 的準備

​SQS Overflow Storage

​建立 Job 類別

​make:job 指令

​Job 類別的結構

​Job 的分派

​dispatch()

​延遲 dispatch

​dispatchAfterResponse()

​分派到特定 queue

​Queue Routing

​Queue 轉發（Queue::forward()）

​同步執行（用於測試／開發）

​大量 dispatch

​Job Chain

​Job Batch

​Job 的處理

​queue:work 指令

​Queue Worker 的監控選項

​在 Job 類別中設定重試

​將崩潰計為例外

​以 Job Middleware 控制執行

​速率限制（RateLimited）

​防止重疊執行（WithoutOverlapping）

​抑制連續發生的例外（ThrottlesExceptions）

​Job 的 Release（Release middleware）

​失敗 Job 的處理

​準備 failed_jobs 資料表

​失敗時的收尾

​依例外停止重試

​檢視失敗 job 清單

​重試失敗 job

​刪除失敗 job

​常用的 queue driver

​database driver

​redis driver

​以 Supervisor 進行正式運行

​實務範例：以 queue 處理寄信

​總結

什麼是 Queue

Queue 的設定

config/queue.php

.env 設定

資料庫 driver 的準備

Redis driver 的準備

SQS Overflow Storage

建立 Job 類別

make:job 指令

Job 類別的結構

Job 的分派

dispatch()

延遲 dispatch

dispatchAfterResponse()

分派到特定 queue

Queue Routing

Queue 轉發（`Queue::forward()`）

同步執行（用於測試／開發）

大量 dispatch

Job Chain

Job Batch

Job 的處理

queue:work 指令

Queue Worker 的監控選項

在 Job 類別中設定重試

將崩潰計為例外

以 Job Middleware 控制執行

速率限制（RateLimited）

防止重疊執行（WithoutOverlapping）

抑制連續發生的例外（ThrottlesExceptions）

Job 的 Release（Release middleware）

失敗 Job 的處理

準備 failed_jobs 資料表

失敗時的收尾

依例外停止重試

檢視失敗 job 清單

重試失敗 job

刪除失敗 job

常用的 queue driver

database driver

redis driver

以 Supervisor 進行正式運行

實務範例：以 queue 處理寄信

總結