佇列 - Queue

• 簡介

  • Connections vs. Queues
  • Driver Notes and Prerequisites

• 建立 Job

  • 產生 Job 類別
  • 類別架構
  • 不重複 Job
  • 加密的 Job

• Job Middleware

  • 頻率限制
  • 避免 Job 重疊
  • 頻率限制的 Exception

• 分派 Job

  • 延遲分派
  • 同步分派
  • Job 與資料庫 Transaction
  • 串聯 Job
  • Customizing The Queue and Connection
  • 指定 Job 最大嘗試次數與逾時
  • 錯誤處理

• 批次 Job

  • 定義批次 Job
  • 分派批次
  • Chains and Batches
  • Adding Jobs to Batches
  • 檢查批次
  • 取消批次
  • 批次執行失敗
  • 修剪批次
  • Storing Batches in DynamoDB

• 將閉包放入佇列

• Running the Queue Worker

  • queue:work 指令
  • 佇列的優先度
  • Queue Workers and Deployment
  • Job Expirations and Timeouts

• Supervisor 設定

• 處理失敗的 Job

  • 在 Job 失敗後進行清理
  • 重試失敗的 Job
  • 忽略不存在的 Model
  • 修剪失敗的 Job
  • Storing Failed Jobs in DynamoDB
  • 禁用失敗 Job 的保存
  • 失敗 Job 事件

• 在佇列中清除 Job

• 監控佇列

• 測試

  • Faking a Subset of Jobs
  • 測試串連的 Job
  • 測試批次的 Job

• Job 事件

簡介

在製作 Web App 時，有些任務若在 Web Request 中進行會花費太多時間，如解析 CSV 檔並上傳。所幸，在 Laravel 中，要建立在背景執行的佇列任務非常輕鬆。只要將需要花費時間的任務移到佇列中執行，就能加速網站對 Request 的回應速度，並提供更好的使用者經驗給客戶。

Laravel 的佇列為各種不同的佇列後端都提供了統一的 API。這些後端包括 Amazon SQS、Redis、甚至是關聯式資料庫。

Laravel 的佇列設定選項保存在專案的 config/queue.php 設定檔中。在這個檔案內，可以看到供各個 Laravel 內建佇列 Driver 使用的連線設定，包含資料庫、Amazon SQS、Redis、Beanstalkd 等 Driver，還包括一個會即時執行任務的同步佇列 (用於本機開發)。還包含一個 null 佇列，用於忽略佇列任務。

Connections vs. Queues

在開始使用 Laravel 佇列前，我們需要先瞭解「連線(Connection)」與「佇列(Queue)」的差別。在 config/queue.php 設定檔中有個 connections 設定陣列。該選項用於定義連到後端佇列服務的連線，後端佇列服務就是像 Amazon SQS、Beanstalk、Redis 等。不過，一個佇列連線可以有多個「佇列」，我們可以將這些不同的佇列想成是幾個不同堆疊的佇列任務。

可以注意到範例 queue 設定檔中的各個範例連線設定中都包含了一個 queue 屬性。這個 queue 屬性指定的，就是當我們將任務傳給這個連線時預設會被分派的佇列。換句話說，若我們在分派任務時沒有顯式定義要分派到哪個佇列上，這個任務就會被分派到連線設定中 queue 屬性所定義的佇列上：

1use App\Jobs\ProcessPodcast;
2 
3// This job is sent to the default connection's default queue...
4ProcessPodcast::dispatch();
5 
6// This job is sent to the default connection's "emails" queue...
7ProcessPodcast::dispatch()->onQueue('emails');
1use App\Jobs\ProcessPodcast;
2 
3// This job is sent to the default connection's default queue...
4ProcessPodcast::dispatch();
5 
6// This job is sent to the default connection's "emails" queue...
7ProcessPodcast::dispatch()->onQueue('emails');

有的程式沒有要將任務推送到不同佇列的需求，這些程式只需要有單一佇列就好了。不過，因為 Laravel 的 Queue Worker 可調整各個 Queue 的優先處理等級，因此如果想要調整不同任務的優先處理順序，把任務推送到不同佇列就很有用。就來來說，我們如果把任務推送到 high 佇列，我們就可以執行一個 Worker 來讓這個佇列以更高優先級處理：

1php artisan queue:work --queue=high,default
1php artisan queue:work --queue=high,default

Driver Notes and Prerequisites

Database

若要使用 database 佇列 Driver，我們需要先有一個用來存放任務的資料庫資料表。若要產生一個用於建立這個資料表的 Migration，請執行 queue:table Artisan 指令。建立好 Migration 後，就可以使用 migrate 指令來 Migrate 資料庫：

1php artisan queue:table
2 
3php artisan migrate
1php artisan queue:table
2 
3php artisan migrate

最後，別忘了更新專案 .env 檔中的 QUEUE_CONNECTION 變數來讓專案使用 database Driver：

1QUEUE_CONNECTION=database
1QUEUE_CONNECTION=database

Redis

若要使用 redis 佇列 Driver，請在 config/database.php 設定檔中設定 Redis 資料庫連線。

Redis Cluster

若 Redis 佇列要使用 Redis Cluster，則設定的佇列名稱必須包含一個 Key Hash Tag。必須加上 Key Hash Tag，這樣才能確保給定佇列中所有的 Redis 索引鍵都有被放在相同的 Hash Slot 中：

1'redis' => [
2    'driver' => 'redis',
3    'connection' => 'default',
4    'queue' => '{default}',
5    'retry_after' => 90,
6],
1'redis' => [
2    'driver' => 'redis',
3    'connection' => 'default',
4    'queue' => '{default}',
5    'retry_after' => 90,
6],

Blocking

在使用 Redis 佇列時，可使用 block_for 設定選項來指定 Redis Driver 在迭代 Worker 迴圈並重新讀取 Redis 資料庫來等新 Job 進來時要等待多久。

可依據佇列的負載來調整這個值，以避免不斷讀取 Redis 資料庫來尋找新任務，會比較有效率。舉例來說，我們可以將其設為 5，表示 Redis Driver 在等待新任務出現時應先等待 5 秒再查詢 Redis 資料庫：

1'redis' => [
2    'driver' => 'redis',
3    'connection' => 'default',
4    'queue' => 'default',
5    'retry_after' => 90,
6    'block_for' => 5,
7],
1'redis' => [
2    'driver' => 'redis',
3    'connection' => 'default',
4    'queue' => 'default',
5    'retry_after' => 90,
6    'block_for' => 5,
7],

其他 Driver 的前置要求

下列 Queue Driver 還需要一些相依性套件。可以使用 Composer 套件管理員來安裝這些相依性套件：

建立 Job

產生 Job 類別

預設情況下，專案中所有可放入佇列的任務都存放在 app/Jobs 目錄內。若 app/Jobs 目錄不存在，則執行 make:jobs Artisan 指令時會建立該目錄：

1php artisan make:job ProcessPodcast
1php artisan make:job ProcessPodcast

產生的類別會實作 Illuminate\Contracts\Queue\ShouldQueue 介面，這樣 Laravel 就知道該 Job 要被推入佇列並以非同步方式執行。

類別架構

Job 類別非常簡單，通常只包含了一個 handle 方法，會在佇列處理 Job 時叫用。要開始使用 Job，我們先來看一個範例 Job 類別。在這個範例中，我們先假裝時我們在管理一個 Podcast 上架服務，我們需要在上架前處理上傳的 Podcast 檔案：

 1<?php
 2 
 3namespace App\Jobs;
 4 
 5use App\Models\Podcast;
 6use App\Services\AudioProcessor;
 7use Illuminate\Bus\Queueable;
 8use Illuminate\Contracts\Queue\ShouldQueue;
 9use Illuminate\Foundation\Bus\Dispatchable;
10use Illuminate\Queue\InteractsWithQueue;
11use Illuminate\Queue\SerializesModels;
12 
13class ProcessPodcast implements ShouldQueue
14{
15    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
16 
17    /**
18     * Create a new job instance.
19     */
20    public function __construct(
21        public Podcast $podcast,
22    ) {}
23 
24    /**
25     * Execute the job.
26     */
27    public function handle(AudioProcessor $processor): void
28    {
29        // Process uploaded podcast...
30    }
31}
 1<?php
 2 
 3namespace App\Jobs;
 4 
 5use App\Models\Podcast;
 6use App\Services\AudioProcessor;
 7use Illuminate\Bus\Queueable;
 8use Illuminate\Contracts\Queue\ShouldQueue;
 9use Illuminate\Foundation\Bus\Dispatchable;
10use Illuminate\Queue\InteractsWithQueue;
11use Illuminate\Queue\SerializesModels;
12 
13class ProcessPodcast implements ShouldQueue
14{
15    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
16 
17    /**
18     * Create a new job instance.
19     */
20    public function __construct(
21        public Podcast $podcast,
22    ) {}
23 
24    /**
25     * Execute the job.
26     */
27    public function handle(AudioProcessor $processor): void
28    {
29        // Process uploaded podcast...
30    }
31}

在這個範例中，可以看到我們直接將 [Eloquent Model] 傳入佇列任務的 Constructor(建構函式) 中。由於該任務有使用 SerializesModels Trait，所以 Eloquent Model 與已載入的關聯 Model 都會被序列化處理，並在處理任務時反序列化。

若佇列任務的 Constructor 中接受 Eloquent Model，則只有 Model 的識別元(Identifier)會被序列化進佇列中。實際要處理任務時，佇列系統會自動從資料庫中重新取得完整的 Model 實體以及已載入的關聯。通過這種序列化 Model 的做法，我們就能縮小傳入佇列 Driver 的任務承載(Payload)。

handle 方法的相依性插入

佇列在處理任務時會叫用該任務的 handle 方法。請注意，我們可以在任務的 handle 方法上型別提示(Type-Hint)任何相依性項目。Laravel Service Container 會自動插入這些相依性。

若想完整控制 Container 要如何插入這些相依性到 handle 方法，可使用 Container 的 bindMethod 方法。bindMethod 方法接收一個回呼，該回呼則接收該任務與 Container。我們可以在這個回呼中自行叫用 handle 方法。一般來說，我們應該從 App\Providers\AppServiceProvider Service Provider 的 boot 方法中叫用這個方法：

1use App\Jobs\ProcessPodcast;
2use App\Services\AudioProcessor;
3use Illuminate\Contracts\Foundation\Application;
4 
5$this->app->bindMethod([ProcessPodcast::class, 'handle'], function (ProcessPodcast $job, Application $app) {
6    return $job->handle($app->make(AudioProcessor::class));
7});
1use App\Jobs\ProcessPodcast;
2use App\Services\AudioProcessor;
3use Illuminate\Contracts\Foundation\Application;
4 
5$this->app->bindMethod([ProcessPodcast::class, 'handle'], function (ProcessPodcast $job, Application $app) {
6    return $job->handle($app->make(AudioProcessor::class));
7});

佇列中的關聯

由於所有已載入的 Eloquent Model 關聯都會在 Job 被放入佇列時序列化，因此，序列化的 Job 字串有時候會比較大。此外，當任務被反序列化，然後 Model 關聯被從資料庫中重新取出時，這些關聯的資料會以完整的關聯取出。這表示，若在 Model 被任務佇列序列化前有對關聯套用任何查詢條件，在反序列化時，這些條件都不會被套用。因此，若只想處理給定關聯中的一部分，應在佇列任務中重新套用這些查詢條件。

或者，若要防止關聯被序列化，可在設定屬性值時在 Model 上呼叫 withoutRelations 方法。該方法會回傳該 Model 不含已載入關聯的實體：

1/**
2 * Create a new job instance.
3 */
4public function __construct(Podcast $podcast)
5{
6    $this->podcast = $podcast->withoutRelations();
7}
1/**
2 * Create a new job instance.
3 */
4public function __construct(Podcast $podcast)
5{
6    $this->podcast = $podcast->withoutRelations();
7}

If you are using PHP constructor property promotion and would like to indicate that an Eloquent model should not have its relations serialized, you may use the WithoutRelations attribute:

 1use Illuminate\Queue\Attributes\WithoutRelations;
 2 
 3/**
 4 * Create a new job instance.
 5 */
 6public function __construct(
 7    #[WithoutRelations]
 8    public Podcast $podcast
 9) {
10}
 1use Illuminate\Queue\Attributes\WithoutRelations;
 2 
 3/**
 4 * Create a new job instance.
 5 */
 6public function __construct(
 7    #[WithoutRelations]
 8    public Podcast $podcast
 9) {
10}

If a job receives a collection or array of Eloquent models instead of a single model, the models within that collection will not have their relationships restored when the job is deserialized and executed. This is to prevent excessive resource usage on jobs that deal with large numbers of models.

不重複 Job

有時候，我們可能會想確保某個任務在佇列中一次只能有一個實體。我們可以在 Job 類別上實作 ShouldBeUnique 介面來確保一次只執行一個實體。要實作這個介面，我們需要在 Class 上定義幾個額外的方法：

1<?php
2 
3use Illuminate\Contracts\Queue\ShouldQueue;
4use Illuminate\Contracts\Queue\ShouldBeUnique;
5 
6class UpdateSearchIndex implements ShouldQueue, ShouldBeUnique
7{
8    ...
9}
1<?php
2 
3use Illuminate\Contracts\Queue\ShouldQueue;
4use Illuminate\Contracts\Queue\ShouldBeUnique;
5 
6class UpdateSearchIndex implements ShouldQueue, ShouldBeUnique
7{
8    ...
9}

在上述範例中，UpdateSearchIndex Job 是不重複(Unique)的。所以，若佇列中已經有該 Job 的另一個實體且尚未執行完畢，就不會再次分派該 Job。

在某些情況下，我們可能會想指定要用來判斷 Job 是否重複的「索引鍵」，或是我們可能會想指定一個逾時時間，讓這個 Job 在執行超過該逾時後就不再判斷是否重複。為此，可在 Job 類別上定義 uniqueId 與 uniqueFor 屬性或方法：

 1<?php
 2 
 3use App\Models\Product;
 4use Illuminate\Contracts\Queue\ShouldQueue;
 5use Illuminate\Contracts\Queue\ShouldBeUnique;
 6 
 7class UpdateSearchIndex implements ShouldQueue, ShouldBeUnique
 8{
 9    /**
10     * The product instance.
11     *
12     * @var \App\Product
13     */
14    public $product;
15 
16    /**
17     * The number of seconds after which the job's unique lock will be released.
18     *
19     * @var int
20     */
21    public $uniqueFor = 3600;
22 
23    /**
24     * Get the unique ID for the job.
25     */
26    public function uniqueId(): string
27    {
28        return $this->product->id;
29    }
30}
 1<?php
 2 
 3use App\Models\Product;
 4use Illuminate\Contracts\Queue\ShouldQueue;
 5use Illuminate\Contracts\Queue\ShouldBeUnique;
 6 
 7class UpdateSearchIndex implements ShouldQueue, ShouldBeUnique
 8{
 9    /**
10     * The product instance.
11     *
12     * @var \App\Product
13     */
14    public $product;
15 
16    /**
17     * The number of seconds after which the job's unique lock will be released.
18     *
19     * @var int
20     */
21    public $uniqueFor = 3600;
22 
23    /**
24     * Get the unique ID for the job.
25     */
26    public function uniqueId(): string
27    {
28        return $this->product->id;
29    }
30}

在上述範例中，UpdateSearchIndex Job 使用 Product ID 來判斷是否重複。因此，若新分派的 Job 有相同的 Product ID，則直到現存 Job 執行完畢前，這個 Job 都會被忽略。此外，若現有的 Job 在一個小時內都未被處理，這個不重複鎖定會被解除，之後若有另一個具相同重複索引鍵的 Job 將可被分派進佇列中。

在開始處理 Job 後仍維持讓 Job 不重複

預設情況下，不重複的 Job 會在執行完成或所有嘗試都失敗後「解除鎖定」。不過，有的情況下，我們可能會想在執行完成前就先解除鎖定 Job。為此，不要在該 Job 上實作 ShouldBeUnique，而是實作 ShouldBeUniqueUntillProcessing Contract：

 1<?php
 2 
 3use App\Models\Product;
 4use Illuminate\Contracts\Queue\ShouldQueue;
 5use Illuminate\Contracts\Queue\ShouldBeUniqueUntilProcessing;
 6 
 7class UpdateSearchIndex implements ShouldQueue, ShouldBeUniqueUntilProcessing
 8{
 9    // ...
10}
 1<?php
 2 
 3use App\Models\Product;
 4use Illuminate\Contracts\Queue\ShouldQueue;
 5use Illuminate\Contracts\Queue\ShouldBeUniqueUntilProcessing;
 6 
 7class UpdateSearchIndex implements ShouldQueue, ShouldBeUniqueUntilProcessing
 8{
 9    // ...
10}

不重複 Job 的鎖定

當分派 ShouldBeUnique 時，Laravel 會在幕後使用 uniqueId 索引鍵來取得一個 Lock(鎖定)。若未能取得 Lock，就不會分派該 Job。當 Job 完成處理或所有嘗試都失敗後，就會解除該 Lock。預設情況下，Laravel 會使用預設的快取 Driver 來取得該 Lock。不過，若想使用其他 Driver 來取得 Lock，可定義一個 uniqueVia 方法，並在該方法中回傳要使用的快取 Driver：

 1use Illuminate\Contracts\Cache\Repository;
 2use Illuminate\Support\Facades\Cache;
 3 
 4class UpdateSearchIndex implements ShouldQueue, ShouldBeUnique
 5{
 6    ...
 7 
 8    /**
 9     * Get the cache driver for the unique job lock.
10     */
11    public function uniqueVia(): Repository
12    {
13        return Cache::driver('redis');
14    }
15}
 1use Illuminate\Contracts\Cache\Repository;
 2use Illuminate\Support\Facades\Cache;
 3 
 4class UpdateSearchIndex implements ShouldQueue, ShouldBeUnique
 5{
 6    ...
 7 
 8    /**
 9     * Get the cache driver for the unique job lock.
10     */
11    public function uniqueVia(): Repository
12    {
13        return Cache::driver('redis');
14    }
15}

加密的 Job

Laravel allows you to ensure the privacy and integrity of a job's data via encryption. To get started, simply add the ShouldBeEncrypted interface to the job class. Once this interface has been added to the class, Laravel will automatically encrypt your job before pushing it onto a queue:

1<?php
2 
3use Illuminate\Contracts\Queue\ShouldBeEncrypted;
4use Illuminate\Contracts\Queue\ShouldQueue;
5 
6class UpdateSearchIndex implements ShouldQueue, ShouldBeEncrypted
7{
8    // ...
9}
1<?php
2 
3use Illuminate\Contracts\Queue\ShouldBeEncrypted;
4use Illuminate\Contracts\Queue\ShouldQueue;
5 
6class UpdateSearchIndex implements ShouldQueue, ShouldBeEncrypted
7{
8    // ...
9}

Job Middleware

使用 Job Middleware 就能讓我們將佇列 Job 包裝在一組自定邏輯內執行，讓我們能減少在各個 Job 內撰寫重複的程式碼。舉例來說，假設有下列這個 handle 方法，該方法會使用 Laravel 的 Redis 頻率限制功能，限制每 5 秒只能處理 1 個 Job：

 1use Illuminate\Support\Facades\Redis;
 2 
 3/**
 4 * Execute the job.
 5 */
 6public function handle(): void
 7{
 8    Redis::throttle('key')->block(0)->allow(1)->every(5)->then(function () {
 9        info('Lock obtained...');
10 
11        // Handle job...
12    }, function () {
13        // Could not obtain lock...
14 
15        return $this->release(5);
16    });
17}
 1use Illuminate\Support\Facades\Redis;
 2 
 3/**
 4 * Execute the job.
 5 */
 6public function handle(): void
 7{
 8    Redis::throttle('key')->block(0)->allow(1)->every(5)->then(function () {
 9        info('Lock obtained...');
10 
11        // Handle job...
12    }, function () {
13        // Could not obtain lock...
14 
15        return $this->release(5);
16    });
17}

雖然我們確實可以這樣寫，但這樣一來 handle 方法的實作就變得很亂，因為我們的程式碼跟 Redis 頻率限制的邏輯混在一起了。此外，這樣的頻率限制邏輯一定也會與其他我們想要作頻率限制的 Job 重複。

我們可以定義一個 Job Middleware 來處理頻率限制，而不用在 handle 方法內處理。Laravel 中沒有預設放置 Job Middleware 的地方，因此我們可以隨意在專案內放置這些 Job Middleware。舉例來說，我們可以把 Middleware 放在 app/Jobs/Middleware 目錄下：

 1<?php
 2 
 3namespace App\Jobs\Middleware;
 4 
 5use Closure;
 6use Illuminate\Support\Facades\Redis;
 7 
 8class RateLimited
 9{
10    /**
11     * Process the queued job.
12     *
13     * @param  \Closure(object): void  $next
14     */
15    public function handle(object $job, Closure $next): void
16    {
17        Redis::throttle('key')
18                ->block(0)->allow(1)->every(5)
19                ->then(function () use ($job, $next) {
20                    // Lock obtained...
21 
22                    $next($job);
23                }, function () use ($job) {
24                    // Could not obtain lock...
25 
26                    $job->release(5);
27                });
28    }
29}
 1<?php
 2 
 3namespace App\Jobs\Middleware;
 4 
 5use Closure;
 6use Illuminate\Support\Facades\Redis;
 7 
 8class RateLimited
 9{
10    /**
11     * Process the queued job.
12     *
13     * @param  \Closure(object): void  $next
14     */
15    public function handle(object $job, Closure $next): void
16    {
17        Redis::throttle('key')
18                ->block(0)->allow(1)->every(5)
19                ->then(function () use ($job, $next) {
20                    // Lock obtained...
21 
22                    $next($job);
23                }, function () use ($job) {
24                    // Could not obtain lock...
25 
26                    $job->release(5);
27                });
28    }
29}

就像這樣，跟 Route Middleware 很像，Job Middleware 會收到正在處理的 Job，以及要繼續執行 Job 時要叫用的回呼。

建立好 Job Middleware 後，我們就可以在 Job 的 middleware 方法內將這個 Middleware 附加上去了。make:job 產生的空 Job 不包含 middleware 方法，所以我們需要手動在 Job 類別中新增這個方法：

 1use App\Jobs\Middleware\RateLimited;
 2 
 3/**
 4 * Get the middleware the job should pass through.
 5 *
 6 * @return array<int, object>
 7 */
 8public function middleware(): array
 9{
10    return [new RateLimited];
11}
 1use App\Jobs\Middleware\RateLimited;
 2 
 3/**
 4 * Get the middleware the job should pass through.
 5 *
 6 * @return array<int, object>
 7 */
 8public function middleware(): array
 9{
10    return [new RateLimited];
11}

頻率限制

雖然我們已經示範了要如何自行撰寫頻率限制的 Job Middleware。不過，其實 Laravel 有內建用來為 Job 做頻率限制的 Middleware。就跟 Route 的 Rate Limiter 一樣，可以使用 RateLimiter Facade 的 for 方法來定義 Job 的頻率限制。

舉例來說，我們可能會想讓使用者能備份資料，而一般的使用者限制為每小時可備份一次，VIP 使用者則不限次數。若要做這種頻率限制，可以在 AppServiceProvider 中的 boot 方法內定義一個 RateLimiter：

 1use Illuminate\Cache\RateLimiting\Limit;
 2use Illuminate\Support\Facades\RateLimiter;
 3 
 4/**
 5 * Bootstrap any application services.
 6 */
 7public function boot(): void
 8{
 9    RateLimiter::for('backups', function (object $job) {
10        return $job->user->vipCustomer()
11                    ? Limit::none()
12                    : Limit::perHour(1)->by($job->user->id);
13    });
14}
 1use Illuminate\Cache\RateLimiting\Limit;
 2use Illuminate\Support\Facades\RateLimiter;
 3 
 4/**
 5 * Bootstrap any application services.
 6 */
 7public function boot(): void
 8{
 9    RateLimiter::for('backups', function (object $job) {
10        return $job->user->vipCustomer()
11                    ? Limit::none()
12                    : Limit::perHour(1)->by($job->user->id);
13    });
14}

在上述範例中，我們定義了一個每小時的頻率限制。除了以小時來定義頻率限制外，也可以使用 perMinute 方法來以分鐘定義頻率限制。此外，我們也可以傳入任意值給頻率限制的 by 方法。傳給 by 的值通常是用來區分不同使用者的：

1return Limit::perMinute(50)->by($job->user->id);
1return Limit::perMinute(50)->by($job->user->id);

定義好頻率限制後，我們就可以使用 Illuminate\Queue\Middleware\RateLimited Middleware 來將這個 Rate Limiter 附加到 Job 上。每當這個 Job 超過頻率限制後，這個 Middleware 就會依照頻率限制的間隔，使用適當的延遲時間來將該 Job 放回到佇列中。

 1use Illuminate\Queue\Middleware\RateLimited;
 2 
 3/**
 4 * Get the middleware the job should pass through.
 5 *
 6 * @return array<int, object>
 7 */
 8public function middleware(): array
 9{
10    return [new RateLimited('backups')];
11}
 1use Illuminate\Queue\Middleware\RateLimited;
 2 
 3/**
 4 * Get the middleware the job should pass through.
 5 *
 6 * @return array<int, object>
 7 */
 8public function middleware(): array
 9{
10    return [new RateLimited('backups')];
11}

將受頻率限制的 Job 放回佇列後，一樣會增加 Job 的 attemps 總數。若有需要可以在 Job 類別上適當地設定 tries 與 maxExceptions 屬性。或者，也可以使用 retryUntil 方法 來定義不再重新嘗試 Job 的時間。

若不想讓 Job 在遇到頻率限制後重新嘗試，可使用 dontRelease 方法：

1/**
2 * Get the middleware the job should pass through.
3 *
4 * @return array<int, object>
5 */
6public function middleware(): array
7{
8    return [(new RateLimited('backups'))->dontRelease()];
9}
1/**
2 * Get the middleware the job should pass through.
3 *
4 * @return array<int, object>
5 */
6public function middleware(): array
7{
8    return [(new RateLimited('backups'))->dontRelease()];
9}

避免 Job 重疊

Laravel 隨附了一個 Illuminate\Queue\Middleware\WithoutOverlapping Middleware，可讓我們依照任意索引鍵來避免 Job 重疊。使用這個 Middleware 就能避免同一個資源同時被多個佇列 Job 修改。

舉例來說，假設我們有個佇列任務會負責更新使用者的信用分數，而我們想避免兩個更新相同 User ID 的信用分數 Job 重疊。為此，可在 Job 的 middleware 方法中回傳 WithoutOverlapping Middleware：

 1use Illuminate\Queue\Middleware\WithoutOverlapping;
 2 
 3/**
 4 * Get the middleware the job should pass through.
 5 *
 6 * @return array<int, object>
 7 */
 8public function middleware(): array
 9{
10    return [new WithoutOverlapping($this->user->id)];
11}
 1use Illuminate\Queue\Middleware\WithoutOverlapping;
 2 
 3/**
 4 * Get the middleware the job should pass through.
 5 *
 6 * @return array<int, object>
 7 */
 8public function middleware(): array
 9{
10    return [new WithoutOverlapping($this->user->id)];
11}

每當有重疊的 Job，這些 Job 都會被重新放到佇列中。可以指定這些被重新放回佇列的 Job 在重新嘗試前必須等待多久的秒數：

1/**
2 * Get the middleware the job should pass through.
3 *
4 * @return array<int, object>
5 */
6public function middleware(): array
7{
8    return [(new WithoutOverlapping($this->order->id))->releaseAfter(60)];
9}
1/**
2 * Get the middleware the job should pass through.
3 *
4 * @return array<int, object>
5 */
6public function middleware(): array
7{
8    return [(new WithoutOverlapping($this->order->id))->releaseAfter(60)];
9}

若想在 Job 重疊時馬上刪除這些重疊的 Job 來讓這些 Job 不被重試，請使用 dontRelease 方法：

1/**
2 * Get the middleware the job should pass through.
3 *
4 * @return array<int, object>
5 */
6public function middleware(): array
7{
8    return [(new WithoutOverlapping($this->order->id))->dontRelease()];
9}
1/**
2 * Get the middleware the job should pass through.
3 *
4 * @return array<int, object>
5 */
6public function middleware(): array
7{
8    return [(new WithoutOverlapping($this->order->id))->dontRelease()];
9}

WithoutOverlapping Middleware 使用 Laravel 的 Atomic Lock 功能提供。有時候，Job 可能會未預期地失敗或逾時，並可能未正確釋放 Lock。因此，我們可以使用 expireAfter 方法來顯式定義一個 Lock 的有效時間。舉例來說，下列範例會讓 Laravel 在 Job 開始處理的 3 分鐘後釋放 WithoutOverlapping Lock：

1/**
2 * Get the middleware the job should pass through.
3 *
4 * @return array<int, object>
5 */
6public function middleware(): array
7{
8    return [(new WithoutOverlapping($this->order->id))->expireAfter(180)];
9}
1/**
2 * Get the middleware the job should pass through.
3 *
4 * @return array<int, object>
5 */
6public function middleware(): array
7{
8    return [(new WithoutOverlapping($this->order->id))->expireAfter(180)];
9}

在多個 Job 類別中共用 Lock 索引鍵

預設情況下，WithoutOverlapping Middleware 只會防止相同類別的重疊 Job。因此，即使某兩個不同的 Job 可能使用相同的 Lock 索引鍵，仍然無法防止 Job 重疊。不過，可以使用 shared 方法來指定讓 Laravel 在多個 Job 類別間套用同樣的索引鍵：

 1use Illuminate\Queue\Middleware\WithoutOverlapping;
 2 
 3class ProviderIsDown
 4{
 5    // ...
 6 
 7 
 8    public function middleware(): array
 9    {
10        return [
11            (new WithoutOverlapping("status:{$this->provider}"))->shared(),
12        ];
13    }
14}
15 
16class ProviderIsUp
17{
18    // ...
19 
20 
21    public function middleware(): array
22    {
23        return [
24            (new WithoutOverlapping("status:{$this->provider}"))->shared(),
25        ];
26    }
27}
 1use Illuminate\Queue\Middleware\WithoutOverlapping;
 2 
 3class ProviderIsDown
 4{
 5    // ...
 6 
 7 
 8    public function middleware(): array
 9    {
10        return [
11            (new WithoutOverlapping("status:{$this->provider}"))->shared(),
12        ];
13    }
14}
15 
16class ProviderIsUp
17{
18    // ...
19 
20 
21    public function middleware(): array
22    {
23        return [
24            (new WithoutOverlapping("status:{$this->provider}"))->shared(),
25        ];
26    }
27}

頻率限制的 Exception

Laravel 中隨附了一個 Illuminate\Queue\Middleware\ThrottlesExceptions Middleware，能讓我們針對 Exception 做頻率限制。每當有 Job 擲回特定數量的 Exception 時，接下來要再次嘗試執行該 Job 前，必須要等待特定的時間過後才能繼續。對於一些使用了不穩定第三方服務的 Job 來說，特別適合使用這個功能。

舉例來說，假設我們有個使用了第三方 API 的佇列 Job，而這個 Job 會擲回 Exception。若要對 Exception 做頻率限制，可以在 Job 的 middleware 方法內回傳 ThrottlesExceptions Middleware。一般來說，這個 Middleware 應放在實作基於時間的 attempts之 Job 內：

 1use DateTime;
 2use Illuminate\Queue\Middleware\ThrottlesExceptions;
 3 
 4/**
 5 * Get the middleware the job should pass through.
 6 *
 7 * @return array<int, object>
 8 */
 9public function middleware(): array
10{
11    return [new ThrottlesExceptions(10, 5)];
12}
13 
14/**
15 * Determine the time at which the job should timeout.
16 */
17public function retryUntil(): DateTime
18{
19    return now()->addMinutes(5);
20}
 1use DateTime;
 2use Illuminate\Queue\Middleware\ThrottlesExceptions;
 3 
 4/**
 5 * Get the middleware the job should pass through.
 6 *
 7 * @return array<int, object>
 8 */
 9public function middleware(): array
10{
11    return [new ThrottlesExceptions(10, 5)];
12}
13 
14/**
15 * Determine the time at which the job should timeout.
16 */
17public function retryUntil(): DateTime
18{
19    return now()->addMinutes(5);
20}

Middleware 的第一個 Constructor 引數為 Exception 的數量，當 Job 擲回這個數量的 Exception 後就會被限制執行。第二個引數則是當被限制執行後，在繼續執行之前所要等待的分鐘數。在上述的範例中，若 Job 在 5 分鐘內擲回了 10 個 Exception，則 Laravel 會等待 5 分鐘，然後再繼續嘗試執行該 Job。

當 Job 擲回 Exception，但還未達到所設定的 Exception 閥值，則一般情況下會馬上重試 Job。不過，也可以在講 Middleware 附加到 Job 上時呼叫 backoff 方法來指定一個以分鐘為單位的數字，來指定 Job 所要延遲的時間：

 1use Illuminate\Queue\Middleware\ThrottlesExceptions;
 2 
 3/**
 4 * Get the middleware the job should pass through.
 5 *
 6 * @return array<int, object>
 7 */
 8public function middleware(): array
 9{
10    return [(new ThrottlesExceptions(10, 5))->backoff(5)];
11}
 1use Illuminate\Queue\Middleware\ThrottlesExceptions;
 2 
 3/**
 4 * Get the middleware the job should pass through.
 5 *
 6 * @return array<int, object>
 7 */
 8public function middleware(): array
 9{
10    return [(new ThrottlesExceptions(10, 5))->backoff(5)];
11}

這個 Middleware 在內部使用了 Laravel 的快取系統來實作頻率限制，並使用了該 Job 的類別名稱來作為快取的「索引鍵」。可以在講 Middleware 附加到 Job 上時呼叫 by 方法來複寫這個索引鍵。當有多個 Job 都使用了同一個第三方服務時，就很適合使用這個方法來讓這些 Job 都共用相同的頻率限制：

 1use Illuminate\Queue\Middleware\ThrottlesExceptions;
 2 
 3/**
 4 * Get the middleware the job should pass through.
 5 *
 6 * @return array<int, object>
 7 */
 8public function middleware(): array
 9{
10    return [(new ThrottlesExceptions(10, 10))->by('key')];
11}
 1use Illuminate\Queue\Middleware\ThrottlesExceptions;
 2 
 3/**
 4 * Get the middleware the job should pass through.
 5 *
 6 * @return array<int, object>
 7 */
 8public function middleware(): array
 9{
10    return [(new ThrottlesExceptions(10, 10))->by('key')];
11}

分派 Job

寫好 Job 類別後，就可以使用 Job 上的 dispatch 方法來分派該 Job。傳給 dispatch 方法的引數會被傳給 Job 的 Constructor：

 1<?php
 2 
 3namespace App\Http\Controllers;
 4 
 5use App\Http\Controllers\Controller;
 6use App\Jobs\ProcessPodcast;
 7use App\Models\Podcast;
 8use Illuminate\Http\RedirectResponse;
 9use Illuminate\Http\Request;
10 
11class PodcastController extends Controller
12{
13    /**
14     * Store a new podcast.
15     */
16    public function store(Request $request): RedirectResponse
17    {
18        $podcast = Podcast::create(/* ... */);
19 
20        // ...
21 
22        ProcessPodcast::dispatch($podcast);
23 
24        return redirect('/podcasts');
25    }
26}
 1<?php
 2 
 3namespace App\Http\Controllers;
 4 
 5use App\Http\Controllers\Controller;
 6use App\Jobs\ProcessPodcast;
 7use App\Models\Podcast;
 8use Illuminate\Http\RedirectResponse;
 9use Illuminate\Http\Request;
10 
11class PodcastController extends Controller
12{
13    /**
14     * Store a new podcast.
15     */
16    public function store(Request $request): RedirectResponse
17    {
18        $podcast = Podcast::create(/* ... */);
19 
20        // ...
21 
22        ProcessPodcast::dispatch($podcast);
23 
24        return redirect('/podcasts');
25    }
26}

若想要有條件地分派 Job，可使用 dispatchIf 與 dispatchUnless` 方法：

1ProcessPodcast::dispatchIf($accountActive, $podcast);
2 
3ProcessPodcast::dispatchUnless($accountSuspended, $podcast);
1ProcessPodcast::dispatchIf($accountActive, $podcast);
2 
3ProcessPodcast::dispatchUnless($accountSuspended, $podcast);

在新的 Laravel 專案中，預設的 Queue Driver 是 sync Driver。該 Driver 會在目前 Request 中的前景 (Foreground) 同步執行 Job。若想要讓 Job 被真正放進佇列中在背景執行，你需要在專案的 config/queue.php 設定檔中指定一個不同的 Queue Driver。

延遲分派

若不想讓 Job 馬上被 Queue Worker 處理，可在分派 Job 時使用 delay 方法。舉例來說，我們來指定讓一個 Job 在分派的 10 分鐘後才被開始處理：

 1<?php
 2 
 3namespace App\Http\Controllers;
 4 
 5use App\Http\Controllers\Controller;
 6use App\Jobs\ProcessPodcast;
 7use App\Models\Podcast;
 8use Illuminate\Http\RedirectResponse;
 9use Illuminate\Http\Request;
10 
11class PodcastController extends Controller
12{
13    /**
14     * Store a new podcast.
15     */
16    public function store(Request $request): RedirectResponse
17    {
18        $podcast = Podcast::create(/* ... */);
19 
20        // ...
21 
22        ProcessPodcast::dispatch($podcast)
23                    ->delay(now()->addMinutes(10));
24 
25        return redirect('/podcasts');
26    }
27}
 1<?php
 2 
 3namespace App\Http\Controllers;
 4 
 5use App\Http\Controllers\Controller;
 6use App\Jobs\ProcessPodcast;
 7use App\Models\Podcast;
 8use Illuminate\Http\RedirectResponse;
 9use Illuminate\Http\Request;
10 
11class PodcastController extends Controller
12{
13    /**
14     * Store a new podcast.
15     */
16    public function store(Request $request): RedirectResponse
17    {
18        $podcast = Podcast::create(/* ... */);
19 
20        // ...
21 
22        ProcessPodcast::dispatch($podcast)
23                    ->delay(now()->addMinutes(10));
24 
25        return redirect('/podcasts');
26    }
27}

Dispatching After the Response is Sent to the Browser

如果你的網頁伺服器使用 FastCGI，則 dispatchAfterResponse 是另一個分派 Job 的方法。該方法會延遲分派 Job，直到 HTTP Response 被傳回使用者瀏覽器後才開始處理Job。這樣一來，在處理佇列 Job 的同時，使用者就能繼續使用我們的網站。一般來說，這種做法應只用於一些只需花費 1 秒鐘的 Job，如寄送 E-Mail 等。由於這些 Job 會在目前的 HTTP Request 中處理，因此使用這種方式分派 Job 就不需要執行 Queue Worker：

1use App\Jobs\SendNotification;
2 
3SendNotification::dispatchAfterResponse();
1use App\Jobs\SendNotification;
2 
3SendNotification::dispatchAfterResponse();

也可以用 dispatch 分派一個閉包，然後在 dispatch 輔助函式後串上一個 afterResponse 方法來在 HTTP Response 被傳送給瀏覽器後執行這個閉包：

1use App\Mail\WelcomeMessage;
2use Illuminate\Support\Facades\Mail;
3 
4dispatch(function () {
5    Mail::to('[email protected]')->send(new WelcomeMessage);
6})->afterResponse();
1use App\Mail\WelcomeMessage;
2use Illuminate\Support\Facades\Mail;
3 
4dispatch(function () {
5    Mail::to('[email protected]')->send(new WelcomeMessage);
6})->afterResponse();

同步分派

若想馬上分派 Job (即，同步(Synchronous))，則可使用 dispatchSync 方法。在使用這個方法時，所分派的 Job 不會被放入佇列，而會在目前的處理程序中馬上執行：

 1<?php
 2 
 3namespace App\Http\Controllers;
 4 
 5use App\Http\Controllers\Controller;
 6use App\Jobs\ProcessPodcast;
 7use App\Models\Podcast;
 8use Illuminate\Http\RedirectResponse;
 9use Illuminate\Http\Request;
10 
11class PodcastController extends Controller
12{
13    /**
14     * Store a new podcast.
15     */
16    public function store(Request $request): RedirectResponse
17    {
18        $podcast = Podcast::create(/* ... */);
19 
20        // Create podcast...
21 
22        ProcessPodcast::dispatchSync($podcast);
23 
24        return redirect('/podcasts');
25    }
26}
 1<?php
 2 
 3namespace App\Http\Controllers;
 4 
 5use App\Http\Controllers\Controller;
 6use App\Jobs\ProcessPodcast;
 7use App\Models\Podcast;
 8use Illuminate\Http\RedirectResponse;
 9use Illuminate\Http\Request;
10 
11class PodcastController extends Controller
12{
13    /**
14     * Store a new podcast.
15     */
16    public function store(Request $request): RedirectResponse
17    {
18        $podcast = Podcast::create(/* ... */);
19 
20        // Create podcast...
21 
22        ProcessPodcast::dispatchSync($podcast);
23 
24        return redirect('/podcasts');
25    }
26}

Job 與資料庫 Transaction

雖然，在資料庫 Transaction 中分派 Job 是完全 OK 的，但應特別注意 Job 能否被正確執行。當我們在 Transaction 中分派 Job 後，這個 Job 很有可能會在上層 Transaction 被 Commit 前就被 Queue Worker 給執行了。這時候，我們在 Transaction 中對 Model 或資料庫記錄所做出的更改都還未反應到資料庫上。而且，在 Transaction 中做建立的 Model 或資料庫記錄也可能還未出現在資料庫中。

幸好，Laravel 提供了數種方法可解決這個狀況。第一種方法，我們可以在 Queue 連線的設定陣列中設定 after_commit 連線選項：

1'redis' => [
2    'driver' => 'redis',
3    // ...
4    'after_commit' => true,
5],
1'redis' => [
2    'driver' => 'redis',
3    // ...
4    'after_commit' => true,
5],

當 after_commit 設為 true 後，我們就可以在資料庫 Transaction 中分派 Job 了。Laravel 會等到未完成的上層資料庫 Transaction 都被 Commit 後才將 Job 分派出去。不過，當然，若目前沒有正在處理的資料庫 Transaction，這個 Job 會馬上被分派。

若因為 Transaction 中發上 Exception 而造成 Transaction 被 Roll Back(回溯)，則在這個 Transaction 間所分派的 Job 也會被取消。

內嵌指定 Commit 的分派行為

若未將 after_commit 佇列連線選項設為 true，則我們還是可以指定讓某個特定的 Job 在所有已開啟的資料庫 Transaction 都被 Commit 後才被分派。若要這麼做，可在分派動作後串上 afterCommit 方法：

1use App\Jobs\ProcessPodcast;
2 
3ProcessPodcast::dispatch($podcast)->afterCommit();
1use App\Jobs\ProcessPodcast;
2 
3ProcessPodcast::dispatch($podcast)->afterCommit();

同樣地，若 after_commit 選項為 true，則我們也可以馬上分派某個特定的 Job，而不等待資料庫 Transaction 的 Commit：

1ProcessPodcast::dispatch($podcast)->beforeCommit();
1ProcessPodcast::dispatch($podcast)->beforeCommit();

Job 的串聯

通過 Job 串聯，我們就可以指定一組佇列 Job 的清單，在主要 Job 執行成功後才依序執行這組 Job。若按照順序執行的其中一個 Job 執行失敗，則剩下的 Job 都將不被執行。若要執行佇列的 Job 串聯，可使用 Bus Facade 中的 chain 方法。Laravel 的 Command Bus(指令匯流排) 是一個低階的原件，佇列 Job 的分派功能就是使用這個原件製作的：

 1use App\Jobs\OptimizePodcast;
 2use App\Jobs\ProcessPodcast;
 3use App\Jobs\ReleasePodcast;
 4use Illuminate\Support\Facades\Bus;
 5 
 6Bus::chain([
 7    new ProcessPodcast,
 8    new OptimizePodcast,
 9    new ReleasePodcast,
10])->dispatch();
 1use App\Jobs\OptimizePodcast;
 2use App\Jobs\ProcessPodcast;
 3use App\Jobs\ReleasePodcast;
 4use Illuminate\Support\Facades\Bus;
 5 
 6Bus::chain([
 7    new ProcessPodcast,
 8    new OptimizePodcast,
 9    new ReleasePodcast,
10])->dispatch();

除了串聯 Job 類別實體，我們也可以串聯閉包：

1Bus::chain([
2    new ProcessPodcast,
3    new OptimizePodcast,
4    function () {
5        Podcast::update(/* ... */);
6    },
7])->dispatch();
1Bus::chain([
2    new ProcessPodcast,
3    new OptimizePodcast,
4    function () {
5        Podcast::update(/* ... */);
6    },
7])->dispatch();

Chain Connection and Queue

若想指定串聯 Job 的連線與佇列，則可使用 onConnection 與 onQueue 方法。除非佇列 Job 有特別指定不同的連線或佇列，否則，這些方法可用來指定要使用的連線名稱與佇列名稱：

1Bus::chain([
2    new ProcessPodcast,
3    new OptimizePodcast,
4    new ReleasePodcast,
5])->onConnection('redis')->onQueue('podcasts')->dispatch();
1Bus::chain([
2    new ProcessPodcast,
3    new OptimizePodcast,
4    new ReleasePodcast,
5])->onConnection('redis')->onQueue('podcasts')->dispatch();

串聯失敗

將 Job 串聯起來後，可使用 catch 方法來指定當串聯中有 Job 失敗時要被叫用的閉包。給定的回呼會收到一個導致 Job 失敗的 Throwable 實體：

 1use Illuminate\Support\Facades\Bus;
 2use Throwable;
 3 
 4Bus::chain([
 5    new ProcessPodcast,
 6    new OptimizePodcast,
 7    new ReleasePodcast,
 8])->catch(function (Throwable $e) {
 9    // A job within the chain has failed...
10})->dispatch();
 1use Illuminate\Support\Facades\Bus;
 2use Throwable;
 3 
 4Bus::chain([
 5    new ProcessPodcast,
 6    new OptimizePodcast,
 7    new ReleasePodcast,
 8])->catch(function (Throwable $e) {
 9    // A job within the chain has failed...
10})->dispatch();

Customizing The Queue a Connection

Dispatching to a Particular Queue

我們可以將 Job 分門別類放入不同的佇列中，進而分類管理這些 Job，甚至能針對不同佇列設定優先度、指定要有多少個 Worker。不過請記得，放入不同佇列不會將 Job 推送到佇列設定檔中所定義的不同佇列「連線」上，而只會將 Job 推入單一連線中指定的佇列。若要指定佇列，請在分派 Job 時使用 onQueue 方法：

 1<?php
 2 
 3namespace App\Http\Controllers;
 4 
 5use App\Http\Controllers\Controller;
 6use App\Jobs\ProcessPodcast;
 7use App\Models\Podcast;
 8use Illuminate\Http\RedirectResponse;
 9use Illuminate\Http\Request;
10 
11class PodcastController extends Controller
12{
13    /**
14     * Store a new podcast.
15     */
16    public function store(Request $request): RedirectResponse
17    {
18        $podcast = Podcast::create(/* ... */);
19 
20        // Create podcast...
21 
22        ProcessPodcast::dispatch($podcast)->onQueue('processing');
23 
24        return redirect('/podcasts');
25    }
26}
 1<?php
 2 
 3namespace App\Http\Controllers;
 4 
 5use App\Http\Controllers\Controller;
 6use App\Jobs\ProcessPodcast;
 7use App\Models\Podcast;
 8use Illuminate\Http\RedirectResponse;
 9use Illuminate\Http\Request;
10 
11class PodcastController extends Controller
12{
13    /**
14     * Store a new podcast.
15     */
16    public function store(Request $request): RedirectResponse
17    {
18        $podcast = Podcast::create(/* ... */);
19 
20        // Create podcast...
21 
22        ProcessPodcast::dispatch($podcast)->onQueue('processing');
23 
24        return redirect('/podcasts');
25    }
26}

或者，也可以在 Job 的 Constructor 中呼叫 onQueue 方法來指定 Job 的佇列：

 1<?php
 2 
 3namespace App\Jobs;
 4 
 5 use Illuminate\Bus\Queueable;
 6 use Illuminate\Contracts\Queue\ShouldQueue;
 7 use Illuminate\Foundation\Bus\Dispatchable;
 8 use Illuminate\Queue\InteractsWithQueue;
 9 use Illuminate\Queue\SerializesModels;
10 
11class ProcessPodcast implements ShouldQueue
12{
13    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
14 
15    /**
16     * Create a new job instance.
17     */
18    public function __construct()
19    {
20        $this->onQueue('processing');
21    }
22}
 1<?php
 2 
 3namespace App\Jobs;
 4 
 5 use Illuminate\Bus\Queueable;
 6 use Illuminate\Contracts\Queue\ShouldQueue;
 7 use Illuminate\Foundation\Bus\Dispatchable;
 8 use Illuminate\Queue\InteractsWithQueue;
 9 use Illuminate\Queue\SerializesModels;
10 
11class ProcessPodcast implements ShouldQueue
12{
13    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
14 
15    /**
16     * Create a new job instance.
17     */
18    public function __construct()
19    {
20        $this->onQueue('processing');
21    }
22}

Dispatching to a Particular Connection

若專案有使用到多個佇列連線，則可以使用 onConnection 方法來指定要將 Job 推送到哪個連線：

 1<?php
 2 
 3namespace App\Http\Controllers;
 4 
 5use App\Http\Controllers\Controller;
 6use App\Jobs\ProcessPodcast;
 7use App\Models\Podcast;
 8use Illuminate\Http\RedirectResponse;
 9use Illuminate\Http\Request;
10 
11class PodcastController extends Controller
12{
13    /**
14     * Store a new podcast.
15     */
16    public function store(Request $request): RedirectResponse
17    {
18        $podcast = Podcast::create(/* ... */);
19 
20        // Create podcast...
21 
22        ProcessPodcast::dispatch($podcast)->onConnection('sqs');
23 
24        return redirect('/podcasts');
25    }
26}
 1<?php
 2 
 3namespace App\Http\Controllers;
 4 
 5use App\Http\Controllers\Controller;
 6use App\Jobs\ProcessPodcast;
 7use App\Models\Podcast;
 8use Illuminate\Http\RedirectResponse;
 9use Illuminate\Http\Request;
10 
11class PodcastController extends Controller
12{
13    /**
14     * Store a new podcast.
15     */
16    public function store(Request $request): RedirectResponse
17    {
18        $podcast = Podcast::create(/* ... */);
19 
20        // Create podcast...
21 
22        ProcessPodcast::dispatch($podcast)->onConnection('sqs');
23 
24        return redirect('/podcasts');
25    }
26}

也可以將 onConnection 與 onQueue 方法串聯在一起來指定 Job 的連線與佇列：

1ProcessPodcast::dispatch($podcast)
2              ->onConnection('sqs')
3              ->onQueue('processing');
1ProcessPodcast::dispatch($podcast)
2              ->onConnection('sqs')
3              ->onQueue('processing');

或者，也可以在 Job 的 Constructor 中呼叫 onConnection 來指定 Job 的連線：

 1<?php
 2 
 3namespace App\Jobs;
 4 
 5 use Illuminate\Bus\Queueable;
 6 use Illuminate\Contracts\Queue\ShouldQueue;
 7 use Illuminate\Foundation\Bus\Dispatchable;
 8 use Illuminate\Queue\InteractsWithQueue;
 9 use Illuminate\Queue\SerializesModels;
10 
11class ProcessPodcast implements ShouldQueue
12{
13    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
14 
15    /**
16     * Create a new job instance.
17     */
18    public function __construct()
19    {
20        $this->onConnection('sqs');
21    }
22}
 1<?php
 2 
 3namespace App\Jobs;
 4 
 5 use Illuminate\Bus\Queueable;
 6 use Illuminate\Contracts\Queue\ShouldQueue;
 7 use Illuminate\Foundation\Bus\Dispatchable;
 8 use Illuminate\Queue\InteractsWithQueue;
 9 use Illuminate\Queue\SerializesModels;
10 
11class ProcessPodcast implements ShouldQueue
12{
13    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
14 
15    /**
16     * Create a new job instance.
17     */
18    public function __construct()
19    {
20        $this->onConnection('sqs');
21    }
22}

指定最大嘗試次數與逾時

最大嘗試次數

若有某個佇列 Job 遇到錯誤，我們通常不會想讓這個 Job 一直重試。因此，Laravel 提供了多種定義 Job 重試次數的方法。

其中一種指定 Job 最大嘗試次數的方法是在 Artisan 指令列中使用 --tries 開關。使用這種方式指定的嘗試次數會套用到所有該 Worker 處理的 Job，除非 Job 上有指定嘗試次數：

1php artisan queue:work --tries=3
1php artisan queue:work --tries=3

若 Job 嘗試了最大嘗試次數，則這個 Job 會被視為是「執行失敗(Failed)」。更多有關處理執行失敗 Job 的資訊，請參考 執行失敗 Job 的說明文件。若提供 --tries=0 給 queue:work 指令，則失敗的 Job 會被無限次數重試。

也可以用另一種更仔細的方法，就是在 Job 類別內定義這個 Job 的最大嘗試次數。若有在 Job 中指定最大嘗試次數，定義在 Job 類別內的次數會比指令列中 --tries 的值擁有更高的優先度：

 1<?php
 2 
 3namespace App\Jobs;
 4 
 5class ProcessPodcast implements ShouldQueue
 6{
 7    /**
 8     * The number of times the job may be attempted.
 9     *
10     * @var int
11     */
12    public $tries = 5;
13}
 1<?php
 2 
 3namespace App\Jobs;
 4 
 5class ProcessPodcast implements ShouldQueue
 6{
 7    /**
 8     * The number of times the job may be attempted.
 9     *
10     * @var int
11     */
12    public $tries = 5;
13}

If you need dynamic control over a particular job's maximum attempts, you may define a tries method on the job:

1/**
2 * Determine number of times the job may be attempted.
3 */
4public function tries(): int
5{
6    return 5;
7}
1/**
2 * Determine number of times the job may be attempted.
3 */
4public function tries(): int
5{
6    return 5;
7}

基於時間的嘗試限制

除了定義 Job 重試多少次要視為失敗以外，也可以限制 Job 嘗試執行的時間長度。這樣一來，在指定的時間範圍內，Job 就可以不斷重試。若要定義最長可重試時間，請在 Job 類別中定義一個 retryUntil 方法。該方法應回傳 DateTime 實體：

1use DateTime;
2 
3/**
4 * Determine the time at which the job should timeout.
5 */
6public function retryUntil(): DateTime
7{
8    return now()->addMinutes(10);
9}
1use DateTime;
2 
3/**
4 * Determine the time at which the job should timeout.
5 */
6public function retryUntil(): DateTime
7{
8    return now()->addMinutes(10);
9}

最大 Exception 數

有時候，我們可能會想讓 Job 可重試多次，但當出現指定數量的未處理 Exception 後，就視為執行失敗 (與直接使用 release 方法釋放 Job 不同)。若要指定未處理 Exception 數量，可在 Job 類別中定義一個 maxExceptions 屬性：

 1<?php
 2 
 3namespace App\Jobs;
 4 
 5use Illuminate\Support\Facades\Redis;
 6 
 7class ProcessPodcast implements ShouldQueue
 8{
 9    /**
10     * The number of times the job may be attempted.
11     *
12     * @var int
13     */
14    public $tries = 25;
15 
16    /**
17     * The maximum number of unhandled exceptions to allow before failing.
18     *
19     * @var int
20     */
21    public $maxExceptions = 3;
22 
23    /**
24     * Execute the job.
25     */
26    public function handle(): void
27    {
28        Redis::throttle('key')->allow(10)->every(60)->then(function () {
29            // Lock obtained, process the podcast...
30        }, function () {
31            // Unable to obtain lock...
32            return $this->release(10);
33        });
34    }
35}
 1<?php
 2 
 3namespace App\Jobs;
 4 
 5use Illuminate\Support\Facades\Redis;
 6 
 7class ProcessPodcast implements ShouldQueue
 8{
 9    /**
10     * The number of times the job may be attempted.
11     *
12     * @var int
13     */
14    public $tries = 25;
15 
16    /**
17     * The maximum number of unhandled exceptions to allow before failing.
18     *
19     * @var int
20     */
21    public $maxExceptions = 3;
22 
23    /**
24     * Execute the job.
25     */
26    public function handle(): void
27    {
28        Redis::throttle('key')->allow(10)->every(60)->then(function () {
29            // Lock obtained, process the podcast...
30        }, function () {
31            // Unable to obtain lock...
32            return $this->release(10);
33        });
34    }
35}

在這個例子中，這個 Job 會在程式無法在 10 秒內取得 Redis Lock 時被釋放，而這個 Job 在此期間最多可嘗試 25 次。不過，若 Job 中有擲回未處理的 Exception，則會被視為是失敗的 Job。

逾時

通常來說，我們知道某個佇列任務大約需要花多少時間執行。因此，在 Laravel 中，我們可以指定一個「逾時」值。預設情況下，逾時值為 60 秒。若 Job 執行超過逾時值所指定的秒數後，負責處理該 Job 的 Worker 就會以錯誤終止執行。一般來說，Worker 會自動由 Server 上設定的 Process Manager 重新開啟。

可在 Artisan 指令列上使用 --timeout 開關來指定 Job 能執行的最大秒數：

1php artisan queue:work --timeout=30
1php artisan queue:work --timeout=30

若 Job 不斷執行逾時超過其最大重試次數，則該 Job 會被標記為執行失敗。

也可以在 Job 類別中定義該 Job 能執行的最大秒數。若有在 Job 上指定逾時，則在 Job 類別上定義的逾時比在指令列上指定的數字擁有更高的優先度：

 1<?php
 2 
 3namespace App\Jobs;
 4 
 5class ProcessPodcast implements ShouldQueue
 6{
 7    /**
 8     * The number of seconds the job can run before timing out.
 9     *
10     * @var int
11     */
12    public $timeout = 120;
13}
 1<?php
 2 
 3namespace App\Jobs;
 4 
 5class ProcessPodcast implements ShouldQueue
 6{
 7    /**
 8     * The number of seconds the job can run before timing out.
 9     *
10     * @var int
11     */
12    public $timeout = 120;
13}

有時候，如 Socket 或連外 HTTP 連線等的 IO Blocking Process 可能不適用所指定的逾時設定。因此，若有使用到這些功能，也請在這些功能的 API 上指定逾時。舉例來說，若使用 Guzzle，則可像這樣指定連線與 Request 的逾時值：

Failing on Timeout

若想讓 Job 在逾時後被標記為執行失敗，可在 Job 類別上定義 $failOnTimeout 屬性：

1/**
2 * Indicate if the job should be marked as failed on timeout.
3 *
4 * @var bool
5 */
6public $failOnTimeout = true;
1/**
2 * Indicate if the job should be marked as failed on timeout.
3 *
4 * @var bool
5 */
6public $failOnTimeout = true;

錯誤處理

若在處理 Job 時有擲回 Exception，則這個 Job 會被自動釋放回佇列中，好讓這個 Job 能被重新嘗試。被釋放會佇列的 Job 會繼續被重試，直到重試次數達到專案上所設定的最大次數。最大重試次數可使用 queue:work Artisan 指令上的 --tries 開關來定義。或者，也可以在 Job 類別上定義最大重試次數。更多有關如何執行 Queue Worker 的資訊可在本文後方找到。

Manually Releasing a Job

有的時候，我們可能會想手動將 Job 釋放會佇列中，好讓這個 Job 能在稍後重試。若要手動釋放 Job，可以呼叫 release 方法：

1/**
2 * Execute the job.
3 */
4public function handle(): void
5{
6    // ...
7 
8    $this->release();
9}
1/**
2 * Execute the job.
3 */
4public function handle(): void
5{
6    // ...
7 
8    $this->release();
9}