SQS FIFOキューをLaravelで扱う：順序保証つきジョブ処理入門
akinori概要
Laravel のキュー機能は sqs ドライバで Amazon SQS をそのまま利用できます。かつては標準で想定されているのが Standard キュー だけで、順序保証と重複排除を備えた FIFO キュー を使おうとすると、そのままでは動きませんでした。FIFO キューへメッセージを送るには MessageGroupId（必須）が必要ですが、昔の SqsQueue::pushRaw() はこれを付与しなかったためです。
この前提は Laravel 12.27.0 で変わりました。同バージョンで MessageGroupId がネイティブ対応になり、ジョブのディスパッチ時に onGroup()、あるいはジョブクラスに messageGroup() メソッドを定義するだけで FIFO キューを扱えるようになっています。自作コネクタを差し込む必要はもうありません。
本記事ではまず SQS FIFO の基礎を整理し、現行の Laravel で推奨される標準 API を使った実装を本筋として解説します。そのうえで、Laravel 12.27 未満を使う場合や内部挙動を理解するための補足として、独自コネクタを自作する従来の方法も残しておきます。
環境
- PHP 8.3 以上 / Laravel 12.27+（13.x 推奨）
aws/aws-sdk-php- SQS FIFO キュー（キュー名は
.fifoで終わる必要あり）
> Laravel 13 は PHP 8.3 以上が必須です。PHP 8.2 でもサポート期間内ではありますが、本線（Laravel 13）とは噛み合わないため、新規に組むなら 8.3 以上を前提にしてください。
SQS FIFO の前提知識
ここは SQS 側の仕様であり、Laravel のバージョンに関係なく変わりません。
- キュー名は
.fifoで終わる必要がある - 順序保証は
MessageGroupId単位。同じグループ ID のメッセージは送信順に 1 件ずつ配信され、異なるグループ ID 同士は並行処理してよい - 重複排除は「コンテンツベース重複排除（本文の SHA-256 を自動利用）」か「
MessageDeduplicationIdの明示指定」の 2 方式。いずれも5分間の重複排除インターバルがある - スループット上限はデフォルト 1 秒あたり 300 メッセージ（バッチで最大 3,000）。超える場合は高スループット FIFO を検討
現行の推奨：標準 API で FIFO を扱う
Laravel 12.27.0 以降は、sqs ドライバがそのまま FIFO キューに対応します。やることは次の 2 つだけです。
config/queue.phpのsqs接続に.fifoキューを設定する- ジョブのディスパッチ時に
onGroup()でグループ ID を指定する（あるいはジョブ側にmessageGroup()を定義する）
1. 接続設定
// config/queue.php
'sqs' => [
'driver' => 'sqs',
'key' => env('AWS_ACCESS_KEY_ID'),
'secret' => env('AWS_SECRET_ACCESS_KEY'),
'prefix' => env('SQS_PREFIX'),
'queue' => env('SQS_QUEUE', 'my-jobs.fifo'),
'suffix' => env('SQS_SUFFIX'),
'region' => env('AWS_DEFAULT_REGION', 'ap-northeast-1'),
'after_commit' => false,
],
キュー名が .fifo で終わっている点を必ず確認してください。重複排除は、キュー側でコンテンツベース重複排除を有効にしておくと、本文の SHA-256 をもとに SQS が自動で判定してくれます。
EC2 のインスタンスロールや ECS のタスクロールで動かす場合は、key/secret を明示せず AWS SDK のデフォルトプロバイダチェーンに任せるほうが安全です（環境変数を空にしておけば SDK が自動解決します）。
2. ディスパッチ時にグループ ID を指定する
<?php
namespace App\Jobs;
use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Bus\Dispatchable;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Queue\SerializesModels;
class SyncOrderStatus implements ShouldQueue
{
use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
public function __construct(public int $orderId, public string $status) {}
public function handle(): void
{
// 同一 orderId のジョブは投入順に1件ずつ処理される
}
}
ディスパッチ側では、順序を保ちたい単位をグループに指定します。
SyncOrderStatus::dispatch($orderId, 'paid')
->onConnection('sqs')
->onGroup("order-{$orderId}");
これにより、同じ注文 ID のイベントは投入順に 1 件ずつ、別の注文 ID 同士は並行に処理されます。
グループ ID をジョブ側に持たせる
ディスパッチのたびに onGroup() を書く代わりに、ジョブクラスに messageGroup() メソッドを定義してグループ決定ロジックを内包させることもできます。
public function messageGroup(): string
{
return "order-{$this->orderId}";
}
こうしておけば、SyncOrderStatus::dispatch($orderId, 'paid')->onConnection('sqs') とするだけで、グループ ID はジョブ自身が決めてくれます。
ワーカーの起動
php artisan queue:work sqs --queue=my-jobs.fifo
補足：Laravel 12.27 未満での自作コネクタ
> ここから先は Laravel 12.27 未満を使っている場合、あるいは「ネイティブ対応が内部で何をやっているか」を理解したい場合向けの補足です。12.27 以降では不要なので、新規実装では前章の標準 API を使ってください。
ネイティブ対応より前の Laravel では、Illuminate\Queue\SqsQueue::pushRaw() が FIFO に必須の MessageGroupId を送っていませんでした。
public function pushRaw($payload, $queue = null, array $options = [])
{
return $this->sqs->sendMessage([
'QueueUrl' => $this->getQueue($queue),
'MessageBody' => $payload,
])->get('MessageId');
}
Standard キュー用の最小実装だったためです。このため、FIFO 用パラメータを付与するキュー実装とコネクタを自作して差し込むのが定石でした。全体像は次のとおりです。
SqsQueueを継承しpushRaw()で FIFO パラメータを付与するSqsFifoQueueを作る- それを生成する
SqsFifoConnectorを作る - サービスプロバイダで
sqs-fifoドライバとして登録する config/queue.phpにsqs-fifo接続を追加する- ジョブ側でグループ ID を指定できるようにする
1. FIFO 対応のキュー実装
<?php
namespace App\Queue;
use Illuminate\Queue\SqsQueue;
class SqsFifoQueue extends SqsQueue
{
public function __construct(
$sqs,
$default,
$prefix = '',
$suffix = '',
$dispatchAfterCommit = false,
protected string $group = 'default',
protected bool $contentBasedDeduplication = false,
) {
parent::__construct($sqs, $default, $prefix, $suffix, $dispatchAfterCommit);
}
public function pushRaw($payload, $queue = null, array $options = [])
{
$message = [
'QueueUrl' => $this->getQueue($queue),
'MessageBody' => $payload,
'MessageGroupId' => $this->resolveGroup($payload, $options),
];
if (! $this->contentBasedDeduplication) {
$message['MessageDeduplicationId'] =
$options['deduplication_id'] ?? hash('sha256', $payload);
}
return $this->sqs->sendMessage($message)->get('MessageId');
}
protected function resolveGroup(string $payload, array $options): string
{
if (! empty($options['group'])) {
return $options['group'];
}
$decoded = json_decode($payload, true);
return $decoded['data']['messageGroupId'] ?? $this->group;
}
}
2. コネクタ
<?php
namespace App\Queue;
use Aws\Sqs\SqsClient;
use Illuminate\Queue\Connectors\ConnectorInterface;
use Illuminate\Support\Arr;
class SqsFifoConnector implements ConnectorInterface
{
public function connect(array $config)
{
if (! empty($config['key']) && ! empty($config['secret'])) {
$config['credentials'] = Arr::only($config, ['key', 'secret', 'token']);
}
return new SqsFifoQueue(
new SqsClient($config),
$config['queue'],
$config['prefix'] ?? '',
$config['suffix'] ?? '',
$config['after_commit'] ?? false,
$config['group'] ?? 'default',
$config['content_based_deduplication'] ?? false,
);
}
}
EC2 のインスタンスロールや ECS のタスクロールで動かす場合は、key/secret を明示せず AWS SDK のデフォルトプロバイダチェーンに任せるほうが安全です。
3. サービスプロバイダで登録
<?php
namespace App\Providers;
use App\Queue\SqsFifoConnector;
use Illuminate\Support\Facades\Queue;
use Illuminate\Support\ServiceProvider;
class QueueServiceProvider extends ServiceProvider
{
public function boot(): void
{
Queue::addConnector('sqs-fifo', fn () => new SqsFifoConnector);
}
}
4. 接続設定
// config/queue.php
'sqs-fifo' => [
'driver' => 'sqs-fifo',
'key' => env('AWS_ACCESS_KEY_ID'),
'secret' => env('AWS_SECRET_ACCESS_KEY'),
'prefix' => env('SQS_PREFIX'),
'queue' => env('SQS_FIFO_QUEUE', 'my-jobs.fifo'),
'region' => env('AWS_DEFAULT_REGION', 'ap-northeast-1'),
'group' => 'default',
'content_based_deduplication' => false,
],
5. ジョブにグループ ID を持たせる
<?php
namespace App\Queue;
trait WithMessageGroup
{
public ?string $messageGroupId = null;
public function assignGroup(string $groupId): static
{
$this->messageGroupId = $groupId;
return $this;
}
}
> 注意： 以前はこのトレイトのメソッドを onGroup() という名前で定義していましたが、Laravel 12.27 以降は onGroup() が標準 API として提供されるため、自作トレイト側の onGroup() は標準 API と名前が衝突します。12.27 以降に上げる予定があるなら、上記のように別名（例では assignGroup()）へ退避させておくか、そもそも標準 API に寄せてください。
SyncOrderStatus::dispatch($orderId, 'paid')
->onConnection('sqs-fifo')
->assignGroup("order-{$orderId}");
確認方法
同一グループに連続投入し、処理順が崩れないことを確認します。
foreach (['paid', 'packed', 'shipped', 'delivered'] as $status) {
SyncOrderStatus::dispatch(1001, $status)
->onConnection('sqs')
->onGroup('order-1001');
}
handle() でステータスと時刻をログに出し、order-1001 が必ず投入順に並ぶことを確認します。重複排除は、同一ペイロードを 5 分以内に 2 回投げ、2 回目が無視されることで確認できます。
注意点
- 順序保証は
MessageGroupId単位。全ジョブを 1 グループに集約すると直列化し、ワーカーを増やしても速くならない - 可視性タイムアウト中は同グループの後続がブロックされる。処理時間に見合った値を設定する
- 重複排除インターバルは 5 分。時間を空けた同一内容の再投入も重複扱いになる。意図的な再投入は
MessageDeduplicationIdをずらす - スループット上限。300 msg/s を超えるなら高スループット FIFO を検討（重複排除スコープが変わるため要件確認）
- コミット後ディスパッチ。DB トランザクションと併用する場合は
after_commitを適切に設定する - フェアキュー。Laravel 12.27 のネイティブ対応は、SQS の Fair Queue 機能（グループ間で処理機会を公平に回す）を Laravel から直接扱えるようにするもの。特定のグループがキューを占有しにくくなる
まとめ
- かつて Laravel の
sqsドライバはMessageGroupIdを送らず FIFO 非対応だったが、Laravel 12.27.0 以降はネイティブ対応になった - 現行版では、
sqs接続に.fifoキューを設定し、ディスパッチ時にonGroup()を付ける（またはジョブにmessageGroup()を定義する）だけで FIFO を扱える。自作コネクタはもう不要 - 自作の
SqsFifoQueue/SqsFifoConnectorが必要なのは Laravel 12.27 未満のみ。その際、独自トレイトのonGroup()は標準 API と名前が衝突する点に注意する - 順序保証は
MessageGroupId単位。順序が必要な最小単位をグループにすることで、順序保証と並列性を両立できる - 可視性タイムアウト・重複排除インターバル・スループット上限という FIFO 固有の制約を理解しておくことが安定運用の前提になる
Sources:
- Use Amazon's new Fair Queue Feature With SQS in Laravel 12.27 - Laravel News
- [[12.x] Add SQS FIFO and fair queue messageGroup() method support · PR #57421 · laravel/framework](https://github.com/laravel/framework/pull/57421)
- Laravel 13 Released: PHP 8.3, Attributes, Laravel AI, and a Smoother Upgrade Path - Laravel News
＼ 最新情報をチェック ／
