SQS を使用した Symfony Messenger Consume Symfony7

対象: Symfony 7.4 / symfony/amazon-sqs-messenger

参考

Messenger: Sync & Queued Message Handling
上記ドキュメントの「Amazon SQS」の節

Messenger は、同期（Sync）で処理する方法と、キューに積んで非同期に処理する方法（Queued Message Handler）があります。キューを使う場合は config/packages/messenger.yaml で transports を設定します。

Symfony アプリケーションを作成

Web アプリケーション（--webapp）として作成します（--webapp を付けないコンソールアプリケーションでも、ほぼ同じです）。

symfony new symfony-messenger-7.x-playground --version="7.4.x" --webapp

パッケージをインストール

composer require symfony/messenger \
    symfony/amazon-sqs-messenger \
    symfony/serializer-pack

--webapp で作成した場合、symfony/webapp-pack が symfony/doctrine-messenger と symfony/serializer-pack を含むため、symfony/messenger は依存として既に入っています。明示的に指定しても問題ありません。
symfony/serializer-pack は symfony/serializer を含みます。
symfony/messenger の Flex レシピを実行すると、次の案内が表示されます。

 symfony/messenger  instructions:

  * You're ready to use the Messenger component. You can define your own message buses
    or start using the default one right now by injecting the message_bus service
    or type-hinting Symfony\Component\Messenger\MessageBusInterface in your code.

  * To send messages to a transport and handle them asynchronously:

    1. Update the MESSENGER_TRANSPORT_DSN env var in .env if needed
       and framework.messenger.transports.async in config/packages/messenger.yaml;
    2. (if using Doctrine) Generate a Doctrine migration bin/console doctrine:migration:diff
       and execute it bin/console doctrine:migration:migrate
    3. Route your message classes to the async transport in config/packages/messenger.yaml.

  * Read the documentation at https://symfony.com/doc/current/messenger.html

Transport（SQS）を設定

DSN

# .env（コミットされるので、実際の認証情報は書かない）
MESSENGER_TRANSPORT_DSN=sqs://sqs.ap-northeast-1.amazonaws.com/123456789012/sample-queue

DSN の形式は公式ドキュメントに次の例があります。

# キュー URL をそのまま DSN にする（GetQueueUrl API の呼び出しを省略できる）
MESSENGER_TRANSPORT_DSN=https://sqs.eu-west-3.amazonaws.com/123456789012/messages
# ローカルのエミュレータなど（sslmode=disable で http 接続）
MESSENGER_TRANSPORT_DSN=sqs://localhost:9494/messages?sslmode=disable

[!WARNING] 公式ドキュメントには access_key / secret_key を DSN のクエリに含める例もありますが、認証情報は DSN や .env に書かないでください。

ローカル: .env.local（Git 管理外）や、AWS の標準的な認証情報（環境変数・プロファイル）を使う

本番: IAM ロールや、Secrets / 実行基盤のシークレット管理から実環境変数として渡す

access_key / secret_key を省略した場合の認証情報の解決順は、AWS SDK（AsyncAws）の仕様に従います。利用する実行基盤で動作を確認してください。

主なオプション（DSN のクエリ、または options で指定）：

オプション	既定値	内容
`auto_setup`	`true`	送受信時にキューを自動作成する
`queue_name`	`messages`	キュー名
`region`	`eu-west-1`	AWS リージョン
`wait_time`	`20`	ロングポーリングの待機秒数
`buffer_size`	`9`	先読みするメッセージ数
`debug`	`false`	`true` で HTTP リクエスト・レスポンスをログ出力する（性能に影響）

[!NOTE] .fifo で終わるキュー名は FIFO キューになります。AmazonSqsFifoStamp で Message group ID と Message deduplication ID を指定します。

messenger.yaml

# config/packages/messenger.yaml
framework:
    messenger:
        transports:
            async:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                serializer: messenger.transport.symfony_serializer # 既定は PHP の serialize。JSON にするには symfony/serializer が必要
                options:
                    auto_setup: false # 本番ではキューを自動作成しない（IAM 権限を最小化できる）

        routing:
            'App\Message\SampleMessage': async

メッセージとハンドラー

<?php
// src/Message/SampleMessage.php
namespace App\Message;

final readonly class SampleMessage
{
    public function __construct(public string $content)
    {
    }
}

<?php
// src/MessageHandler/SampleMessageHandler.php
namespace App\MessageHandler;

use App\Message\SampleMessage;
use Symfony\Component\Messenger\Attribute\AsMessageHandler;

#[AsMessageHandler]
final class SampleMessageHandler
{
    public function __invoke(SampleMessage $message): void
    {
        // メッセージを処理する
    }
}

メッセージを送信するには、MessageBusInterface を注入して dispatch() します。

use Symfony\Component\Messenger\MessageBusInterface;

public function __construct(private readonly MessageBusInterface $bus)
{
}

// ...
$this->bus->dispatch(new SampleMessage('hello'));

Consume

bin/console messenger:consume async -vv

-vv で処理の詳細ログを表示します。
本番では、systemd や supervisor、コンテナのオーケストレーションなどでワーカーを常駐・再起動させます。ワーカーはメモリ使用量などで停止する場合があります（--memory-limit、--time-limit、--limit）。
ハンドラーで例外が発生した場合のリトライ・失敗キューは、公式ドキュメントの「Retries & Failures」を参照してください。