キューの基礎 — 重い処理をリクエストの外に出す

前章で、注文が確定したあとの処理を place() から切り離しました。待ち時間は変わっていません。 リスナーは同期で動くので、メール送信が終わるまでレスポンスが返りません。

この章で、その線の向こう側をリクエストの外へ出します。第1章で挙げた最後の問題が、ここで解決します。

リクエストの外に出すと何が変わるか

いまの流れはこうです。

リクエスト受信 → 注文を作る → メールを送る → レスポンス返却
                              ~~~~~~~~~~~~ 利用者が待っている

キューを挟むと、こうなります。

リクエスト受信 → 注文を作る → やることを記録する → レスポンス返却
                                    ↓
                          ワーカーが拾って実行 → メールを送る

「やることを記録する」だけなら一瞬で終わります。 実際の送信は、別のプロセスがあとから拾って実行します。利用者はそれを待ちません。

前章で挙げたもう 1 つの問題も、ここで解決します。メールサーバーが落ちているとき、同期リスナーは注文 API を 500 にしていました。キューなら、失敗したジョブが記録として残り、あとから実行し直せます。 注文は成功したまま、メールだけを再試行できます。

リスナーをキューへ載せる

変更は 1 行です。

app/Listeners/SendOrderConfirmation.php
use Illuminate\Contracts\Queue\ShouldQueue;

class SendOrderConfirmation implements ShouldQueue
{
    public function handle(OrderPlaced $event): void
    {
        $event->order->user->notify(new OrderConfirmed($event->order));
    }
}

ShouldQueue を実装すると、Laravel はこのリスナーを直接呼ばずキューへ送ります¹。handle() の中身は変わりません。 前章で線を引いておいたので、動かす作業がこれだけで済んでいます。

イベントに SerializesModels を戻す

キューへ送るとき、Laravel はイベントを文字列に変換して保存します。Order モデルがそのまま入ると、都合が悪いことが起きます。

保存した時点の値がまるごと記録されるので、ワーカーが拾うまでのあいだに注文の状態が変わっても、古い値のまま処理されます。データ量も大きくなります。

前章で外した SerializesModels を戻します。

app/Events/OrderPlaced.php
use Illuminate\Queue\SerializesModels;

class OrderPlaced
{
    use SerializesModels;

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

このトレイトは、モデルを主キーだけに畳んで保存し、取り出すときに読み直します²。ワーカーが処理する時点の値が使われます。

復元されたモデルはリレーションを持っていません

主キーで読み直すので、保存する前に読み込んでいたリレーションは失われます。$event->order->user のように書いていると、ワーカーの中でそこから改めてクエリが飛びます。

第8章で Model::preventLazyLoading(! $this->app->isProduction()) を有効にしました。手元やステージングでワーカーを動かすと、この行で例外になります。

テストでは踏みません。この検査は新規作成した直後のモデルを見逃すようになっており、Order::factory()->create() で作ったモデルがそれに当たります。第14章の wasRecentlyCreated が true になるのと同じ状態です。復元されたモデルは「読み出した行」なので見逃されません。 テストが green なのにワーカーだけ落ちる、という形になります。

必要なリレーションは、イベントを作る前に $order->load('user') で読み込んでおくか、ワーカー側で明示的に取り直してください。

削除されたモデルはジョブごと失敗します

読み直す仕組みなので、ジョブが実行されるまでに対象の行が消えていると、モデルが見つからずジョブが失敗します。注文のようにめったに消えないものなら問題になりません。

一時的なデータや、利用者が削除できるものを扱う場合は、public $deleteWhenMissingModels = true; を宣言すると、失敗として記録せず静かに破棄されます²。「対象が消えているなら、やる必要もない」処理に向きます。

書く場所は、キューに載る側のクラスです。 このあと作る Job クラスならその中に、キュー化したリスナーならリスナーの中に書きます。イベントクラスではありません。

ワーカーを動かす

ここまでで、ジョブは記録されるようになりました。まだ誰も実行しません。

php artisan queue:work

このコマンドが、キューを見張って順に処理するプロセスです²。起動しっぱなしにしておきます。

開発中に忘れやすいところです。 ワーカーを起動していないと、注文 API は速くなるのにメールが 1 通も届きません。エラーも出ません。「記録はされているが、誰も拾っていない」状態です。

どこに記録するかを選ぶ

記録先はドライバで決まります。.env の QUEUE_CONNECTION で切り替えます。

ドライバ	記録先	使いどころ
`sync`	どこにも記録しない	その場で実行する。キューを使わないのと同じ
`database`	`jobs` テーブル	小〜中規模。追加のミドルウェアが要らない
`redis`	Redis	規模が大きいとき。取り出しが速い

Laravel 13 の .env.example は QUEUE_CONNECTION=database です。この連載もこのまま進みます。 第3章でマイグレーションを流したとき、jobs テーブルが一緒に作られていました。あのとき「第16章で使う」と書いたのがこれです。

sync は「キューを使わない」設定です。 名前から非同期を連想しますが、逆です。dispatch した時点でその場で実行されます。テスト環境では既定でこれになっており、あとで扱うテストの挙動に関わります。

Redis を選ぶ判断は、ジョブの量と遅延の要求で決まります。database はジョブを 1 件取るたびにテーブルを読み書きするので、件数が増えるとデータベースの負荷になります。最初から Redis を選ぶ必要はありません。 困ってから移せます。

Job クラスを作る

リスナーはイベントに反応する側でした。イベントと関係なく「これをあとでやっておいて」と頼みたいこともあります。 そのときは Job クラスを作ります。

前章で挙げた「倉庫システムへ出荷指示を送る」を書いてみます。

php artisan make:job NotifyWarehouse

app/Jobs/NotifyWarehouse.php
<?php

namespace App\Jobs;

use App\Models\Order;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Queue\Queueable;
use Illuminate\Support\Facades\Http;

class NotifyWarehouse implements ShouldQueue
{
    use Queueable;

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

    public function handle(): void
    {
        Http::post('https://warehouse.example.com/shipments', [
            'order_id' => $this->order->id,
            'items' => $this->order->items->map(fn ($item) => [
                'product_id' => $item->product_id,
                'quantity' => $item->quantity,
            ]),
        ])->throw();
    }
}

投入は dispatch です。どこから投入するかを決める必要があります。 出荷指示も「注文が確定した」に反応するものなので、前章と同じくリスナーを作ります。

app/Listeners/RequestShipment.php
<?php

namespace App\Listeners;

use App\Events\OrderPlaced;
use App\Jobs\NotifyWarehouse;

class RequestShipment
{
    public function handle(OrderPlaced $event): void
    {
        NotifyWarehouse::dispatch($event->order);
    }
}

このリスナーには ShouldQueue を付けていません。 付ける必要がないからです。dispatch はジョブを記録するだけで一瞬で終わります。重い処理は NotifyWarehouse の側にあり、そちらが既に非同期です。 リスナーまでキューに載せると、ジョブを積むためのジョブを積むことになります。

前章で作った SendOrderConfirmation との違いはここです。あちらは handle() の中で直接メールを送るので、リスナー自体を非同期にする必要がありました。

SerializesModels を書いていません。 make:job が付ける Queueable トレイトの中に含まれているためです。イベントのほうには含まれていないので、前節で自分で足しました。ジョブとイベントで書き方が違うのはこの理由です。

->throw() を付けています。 外部 API が 500 を返しても、Http::post() はそれだけでは例外を投げません。付けないと、失敗しているのにジョブが成功扱いで消えます。ジョブの成否は、例外を投げるかどうかで決まります。

リスナーと Job のどちらにするか

判断は「きっかけが何か」です。

「注文が確定した」に反応するもの → リスナー。発表側は聞き手を知らない
「この処理をあとでやる」と名指しで頼むもの → Job。呼ぶ側が対象を知っている

出荷指示は前者なので、リスナーが受けて Job を dispatch する形にしました。リスナーを軽く保ったまま、重い処理を別の単位に切れます。

Job を直接 dispatch するのは、きっかけが明確な場合です。管理画面の「再送」ボタンから NotifyWarehouse::dispatch($order) を呼ぶような場面では、イベントを挟む理由がありません。

トランザクションの中で dispatch しない

第14章で「トランザクションの中でキューへの投入をしない」と書きました。理由をここで回収します。

❌ Bad
DB::transaction(function () use ($user, $items) {
    $order = Order::create([...]);

    NotifyWarehouse::dispatch($order);   // まだコミットされていない

    // ... 明細の作成 ...
});

ワーカーは別のプロセスです。 このトランザクションの中を見られません。コミット前にジョブが記録されると、ワーカーがそれを拾った瞬間にまだ存在しない注文を読みに行きます。SerializesModels が主キーで読み直すので、「見つからない」で失敗します。

タイミング次第で成功したり失敗したりします。手元では再現せず、本番の忙しい時間帯だけ落ちるという形になります。

ロールバックした場合はもっと悪くなります。注文は無かったことになるのに、出荷指示のジョブだけがキューに残ります。

この失敗が起きるかはドライバで変わります

上の説明が当てはまるのは、キューの記録先がアプリケーションのデータベースと別になっているときです。Redis や SQS がこれにあたります。

database ドライバの既定では起きません。config/queue.php の connections.database.connection が未設定 (DB_QUEUE_CONNECTION が空) だと、jobs テーブルへの INSERT が同じ接続を使うため、同じトランザクションに入ります³。コミット前のジョブは他のプロセスから見えず、ロールバックすればジョブも一緒に消えます。

つまりこの連載の既定構成では、上の Bad 例は実際には壊れません。それでもトランザクションの外へ出すのは、ドライバを Redis に替えた瞬間に壊れるからです。壊れ方が構成に依存するコードは、壊れていないうちに直しておくほうが安全です。 移行時に「なぜか本番だけジョブが失敗する」を調べることになります。

解決は 2 通り

1 つは、前章と同じくトランザクションを抜けてから投入することです。位置が目で見えるので、この連載ではこちらを採ります。

もう 1 つは、投入をコミット後まで遅らせる設定です。接続ごとに指定できます²。

config/queue.php
'redis' => [
    'driver' => 'redis',
    // ...
    'after_commit' => true,
],

Redis 側に書いています。 database 接続では、そもそもジョブがトランザクションに乗るので設定しても変わりません。この設定が意味を持つのは、記録先がデータベースの外にあるときです。

呼び出しごとの指定もできます。

NotifyWarehouse::dispatch($order)->afterCommit();

逆に、after_commit を有効にしたうえで特定のジョブだけ即座に投入したいときは beforeCommit() を使います²。

設定で一律に有効化する場合は、チーム全員がそれを知っている必要があります。 コードを読んでも dispatch が遅延することは分かりません。前章の ShouldDispatchAfterCommit と同じトレードオフです。

行き先をまとめて決める

ジョブが増えると、どのジョブをどのキューへ流すかを決めたくなります。メールは軽いので速く回したい、PDF 生成は重いので専用のワーカーに任せたい、といった場面です。

投入時に指定できます。

NotifyWarehouse::dispatch($order)->onQueue('warehouse');

Laravel 13 では、この対応をまとめて宣言できます²。

app/Providers/AppServiceProvider.php
use Illuminate\Support\Facades\Queue;

public function boot(): void
{
    Queue::route(NotifyWarehouse::class, queue: 'warehouse');
}

dispatch する側は行き先を書きません。運用の都合でキューを分けたくなったとき、dispatch を書いた場所を探して回らずに済みます。 接続も指定でき、インターフェースや親クラスに対しても設定できます。

ワーカー側は、担当するキューを指定して起動します。

php artisan queue:work --queue=warehouse

ワーカーは起動したまま動き続ける

第6章で、singleton の注意点として触れた話がここで実物になります。

通常の Web リクエストは、処理が終わればプロセスの状態が捨てられます。ワーカーは違います。 1 つのジョブを処理しても終了せず、次のジョブを取りに行きます。毎回フレームワークを起動し直さないぶん速くなりますが、状態が残ります。

singleton で登録したクラスに前のジョブの情報が残っていると、次のジョブがそれを読みます。第6章の練習問題で扱った請求書番号の例が、まさにこの状況です。別の顧客のデータが混ざります。

ジョブの中で状態を持ち越さないのが基本です。必要な値はジョブのプロパティとして受け取り、処理が終わったら何も残さない形にします。

コードを直してもワーカーは古いまま動きます

ワーカーは起動時にフレームワークを読み込み、そのまま動き続けます。ファイルを書き換えても反映されません。

開発中に「直したのに挙動が変わらない」と感じたら、ワーカーを再起動してください。本番のデプロイでも同じ問題が起きます。その扱いは第17章で詳しく扱います。

テスト

第10章で予告したとおり、Queue::fake() を使います。

ジョブが投入されたことを確認する

tests/Feature/WarehouseNotificationTest.php
use App\Jobs\NotifyWarehouse;
use Illuminate\Foundation\Testing\RefreshDatabase;
use Illuminate\Support\Facades\Queue;
use Laravel\Sanctum\Sanctum;

pest()->use(RefreshDatabase::class);

test('注文すると倉庫への通知ジョブが積まれる', function () {
    Queue::fake();

    Sanctum::actingAs(User::factory()->create());

    $product = Product::factory()->create(['price' => 1200]);
    Stock::factory()->create(['product_id' => $product->id, 'quantity' => 10]);

    $this->postJson('/api/v1/orders', [
        'items' => [['product_id' => $product->id, 'quantity' => 3]],
    ])->assertStatus(201);

    Queue::assertPushed(NotifyWarehouse::class);
});

Queue::fake() はジョブを実行しません。 積まれた記録だけを残します。assertPushed で「何が積まれたか」を確認します。

fake がテストの意味を変えることに注意する

Queue::fake() を呼ぶと、そのテストではジョブの中身が一度も動きません。上のテストは「投入されたこと」しか確認していません。倉庫 API を呼ぶ処理が壊れていても green のままです。

ジョブの中身は、別に直接呼んで確かめます。

tests/Feature/NotifyWarehouseTest.php
use App\Jobs\NotifyWarehouse;
use Illuminate\Foundation\Testing\RefreshDatabase;
use Illuminate\Http\Client\RequestException;
use Illuminate\Support\Facades\Http;

pest()->use(RefreshDatabase::class);

test('倉庫APIが失敗するとジョブも失敗する', function () {
    Http::fake([
        'warehouse.example.com/*' => Http::response('', 500),
    ]);

    $order = Order::factory()->create();

    expect(fn () => (new NotifyWarehouse($order))->handle())
        ->toThrow(RequestException::class);
});

Http::fake() で外部への通信を止めています⁴。第10章で「第16章で外部 API を呼ぶジョブを書くとき使う」と予告したのがこれです。実際に外部へリクエストを飛ばすテストは、相手のサービスに迷惑をかけるうえ、相手が落ちていると自分のテストも落ちます。

このテストは ->throw() を消すと red になります。外部 API の失敗をジョブの失敗として扱えているかを固定できます。

テスト環境では既定でキューを使いません

phpunit.xml は QUEUE_CONNECTION=sync を設定しています。Queue::fake() を呼ばなければ、ジョブはその場で実行されます。

このため、fake() を書かないテストは「非同期にしたつもりが同期で通っている」状態になります。落ちないので気づきません。投入だけを確認したいのか、実行まで確認したいのかを決めてから書いてください。

本番で効く注意点

ワーカーが落ちたら誰も処理しません

queue:work はただのプロセスです。サーバーの再起動でも、例外での異常終了でも止まります。止まったことに誰も気づかないと、ジョブが溜まり続けます。

プロセスを監視して自動的に起動し直す仕組みが要ります。第17章で扱います。

失敗したジョブは記録されますが、誰も見ていません

試行回数を使い切ったジョブは failed_jobs テーブルに記録されます。config/queue.php の failed 設定が既定でそうなっています³。捨てられるわけではないので、あとから再実行できます。

問題は、誰も見ていなければ記録されているだけだという点です。「メールが届かない」という問い合わせを受けて初めて気づく状態では、非同期にした代償だけを払っています。

失敗の通知、再実行の手順、リトライ回数の設計は第17章で扱います。この章の状態のまま本番に出さないでください。

順番は保証されません

複数のワーカーを動かすと、投入した順に処理されるとは限りません。「注文確定メール」より先に「発送完了メール」が届くといったことが起こりえます。

順序が必要なら、ジョブを分けずに 1 つにまとめるか、後続の処理を前の処理の中から投入します。

長く走るジョブに注意する

ワーカーには実行時間の上限があります。上限を超えると途中で止められます。途中まで進んだ処理が中途半端に残ります。

重い処理は小さいジョブに分けます。1 万件の更新なら、100 件ずつ 100 個のジョブにするほうが、失敗したときの影響が小さくなります。

`sync` のまま本番に出さない

QUEUE_CONNECTION=sync は、キューを使わずその場で実行する設定です。本番でこれになっていると、非同期にした意味がまるごと消えます。

しかもエラーは出ません。レスポンスが遅いだけです。デプロイ先の環境変数を確認してください。

まとめ

キューは「やること」を記録し、別のプロセスがあとから実行する仕組み
リスナーは ShouldQueue を実装するだけで非同期になる。handle() は変わらない
queue:work を起動しないと誰も処理しない。エラーも出ずに溜まり続ける
SerializesModels はモデルを主キーに畳んで保存し、実行時に読み直す。行が消えていると失敗する
sync は「キューを使わない」設定。名前に反して同期実行になる
トランザクションの中で dispatch しない。ワーカーは別プロセスなので、まだ存在しない行を読みに行く
ただしこの壊れ方はドライバ次第。database の既定は同じ接続なので壊れず、Redis に替えた瞬間に壊れる
遅らせたいときは after_commit 設定か afterCommit()。ただし読んで分からなくなる
失敗したジョブは failed_jobs に記録される。消えるわけではないが、見る仕組みが無ければ気づけない
Queue::route() で、ジョブごとの行き先を 1 か所にまとめられる
ワーカーは起動したまま動き続ける。状態を持ち越さない。コードを直したら再起動する
Queue::fake() は投入しか確認しない。ジョブの中身は別に直接呼んで確かめる

次に読む

非同期にはなりましたが、本番で運用できる状態ではありません。ワーカーが落ちても誰も気づかず、failed_jobs に記録が積まれても誰も見ません。同じジョブが 2 回実行されることもあります。次章キューを本番で運用する — 冪等性・リトライ・failed_jobs では、失敗の通知と再実行、リトライ回数と待ち時間の設計、ワーカーの監視と再起動を扱います。第13章と第14章で触れた冪等性の、実行側の話もここで回収します。

練習問題

次のコードは、ときどき「注文が見つからない」でジョブが失敗します。原因を説明してください

public function place(User $user, array $items, ?string $idempotencyKey = null): Order
{
    return DB::transaction(function () use ($user, $items, $idempotencyKey) {
        // ... 在庫の減算 ...

        $order = Order::create([...]);

        NotifyWarehouse::dispatch($order);

        foreach ($items as $item) {
            $order->items()->create([...]);
        }

        return $order;
    }, attempts: 3);
}

解答例

コミットする前にジョブを投入しています。

ワーカーは別のプロセスで動いています。このトランザクションの中は見えません。Order::create() で作られた行は、コミットするまで他のプロセスから見えない状態です。

ワーカーがジョブを拾うのがコミットより早いと、SerializesModels が主キーで注文を読み直したときに見つかりません。ジョブが失敗します。タイミングで結果が変わるので、手元では再現せず本番だけで落ちます。

ここで、この症状が出るかどうかは QUEUE_CONNECTION で変わります。

Redis や SQS — 記録先がアプリケーションのデータベースと別なので、上のとおり起きます。ロールバックしてもジョブだけが残ります
database の既定 — jobs への INSERT が同じトランザクションに入るので起きません。ロールバックすればジョブも消えます

「いまは起きないから直さなくてよい」とは言えません。 ドライバを Redis に替えた瞬間に壊れます。そのとき原因を探すのは、キューの移行作業の最中です。構成に依存して壊れるコードは、壊れていないうちに直しておきます。

投入はトランザクションを抜けてから行います。位置を動かせない事情があるなら afterCommit() を使います。

「Queue::fake() を使ったテストが通ったので、メール送信は動いています」という報告にどう応じますか

動いているとは言えません。 Queue::fake() はジョブを実行しないので、確認できたのは「投入されたこと」だけです。

メールを組み立てる処理に誤りがあっても、宛先が間違っていても、外部サービスへの接続設定が誤っていても、そのテストは green のままです。投入と実行は別々に確認する必要があります。

確認するなら 2 つに分けます。投入されることは Queue::fake() と assertPushed で。ジョブの中身は、ジョブを直接 handle() して確かめます。前章で書いた Notification::fake() を使うテストが後者にあたります。

もう 1 つ確認したいことがあります。そのテストは本当に非同期の経路を通ったのかです。テスト環境の QUEUE_CONNECTION は sync なので、Queue::fake() を書き忘れていた場合、ジョブはその場で実行されて通ります。同期のまま通ったのか、投入だけを確認したのかは、テストコードを見ないと分かりません。

そして最後に、ワーカーが動いているかは別の問題です。テストは queue:work の起動を確認しません。デプロイ先でワーカーが起動していなければ、テストが全部通っていてもメールは 1 通も届きません。

出典: Events(Laravel 公式ドキュメント 13.x)。リスナーが ShouldQueue を実装すると、Laravel がそのリスナーをキューへ送ることについて。 ↩
出典: Queues(Laravel 公式ドキュメント 13.x)。次の 6 点について。queue:work がワーカーを起動し --queue で対象を絞れること。SerializesModels がモデルを主キーで復元すること。deleteWhenMissingModels を有効にするとモデルが見つからないジョブを失敗させず破棄すること。接続設定の after_commit と、呼び出し時の afterCommit() / beforeCommit() によるコミット後投入。onQueue() による行き先指定。Queue::route() がジョブクラス・インターフェース・親クラスに対して既定の接続とキューを宣言でき、サービスプロバイダの boot に置くこと。 ↩ ↩² ↩³ ↩⁴ ↩⁵ ↩⁶
出典: config/queue.php(laravel/laravel 13.x)。connections.database.connection が env('DB_QUEUE_CONNECTION') で既定は未設定であること、failed の driver が database-uuids でテーブルが failed_jobs であることについて。 ↩ ↩²
出典: HTTP Client(Laravel 公式ドキュメント 13.x)。Http::fake() による送信の差し替えと、throw() がエラーレスポンスに対して例外を投げることについて。 ↩

リクエストの外に出すと何が変わるか​

リスナーをキューへ載せる​

イベントに SerializesModels を戻す​

ワーカーを動かす​

どこに記録するかを選ぶ​

Job クラスを作る​

リスナーと Job のどちらにするか​

トランザクションの中で dispatch しない​

解決は 2 通り​

行き先をまとめて決める​

ワーカーは起動したまま動き続ける​

テスト​

ジョブが投入されたことを確認する​

fake がテストの意味を変えることに注意する​

本番で効く注意点​

ワーカーが落ちたら誰も処理しません​

失敗したジョブは記録されますが、誰も見ていません​

順番は保証されません​

長く走るジョブに注意する​

sync のまま本番に出さない​

まとめ​

次に読む​

練習問題​

Footnotes​