Symfony Messenger完全解説：キュー、ワーカー、非同期アーキテクチャの面接対策2026

namespace App\Message;

final readonly class InvoiceGenerated
{
    public function __construct(
        public int $orderId,
        public string $customerEmail,
    ) {}
}

namespace App\MessageHandler;

use App\Message\InvoiceGenerated;
use App\Service\InvoiceService;
use App\Service\MailerService;
use Symfony\Component\Messenger\Attribute\AsMessageHandler;

#[AsMessageHandler]
final readonly class InvoiceGeneratedHandler
{
    public function __construct(
        private InvoiceService $invoiceService,
        private MailerService $mailerService,
    ) {}

    public function __invoke(InvoiceGenerated $message): void
    {
        $pdf = $this->invoiceService->generatePdf($message->orderId);
        $this->mailerService->sendInvoice(
            $message->customerEmail,
            $pdf,
        );
    }
}

$this->bus->dispatch(new InvoiceGenerated(
    orderId: $order->getId(),
    customerEmail: $order->getCustomer()->getEmail(),
));

# config/packages/messenger.yaml
framework:
    messenger:
        failure_transport: failed

        transports:
            async_priority_high:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                options:
                    queue_name: high
                retry_strategy:
                    max_retries: 3
                    delay: 1000
                    multiplier: 3
                    max_delay: 60000

            async_priority_low:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                options:
                    queue_name: low
                retry_strategy:
                    max_retries: 5
                    delay: 5000
                    multiplier: 2

            failed:
                dsn: 'doctrine://default?queue_name=failed'

        routing:
            'App\Message\InvoiceGenerated': async_priority_high
            'App\Message\CleanupTempFiles': async_priority_low

# 高優先度メッセージを先に消費し、次に低優先度を処理
php bin/console messenger:consume async_priority_high async_priority_low \
    --memory-limit=128M \
    --time-limit=3600 \
    --limit=500

; /etc/supervisor/conf.d/messenger-worker.conf
[program:messenger-consume]
command=php /var/www/app/bin/console messenger:consume async_priority_high async_priority_low --memory-limit=128M --time-limit=3600
user=www-data
numprocs=2
process_name=%(program_name)s_%(process_num)02d
autostart=true
autorestart=true
startsecs=0
stopwaitsecs=30
stdout_logfile=/var/log/messenger-worker.log
stderr_logfile=/var/log/messenger-worker-error.log

# config/packages/messenger.yaml
framework:
    messenger:
        default_bus: command.bus
        buses:
            command.bus:
                middleware:
                    - validation
                    - doctrine_transaction
            query.bus:
                middleware:
                    - validation
            event.bus:
                default_middleware:
                    allow_no_handlers: true
                middleware:
                    - validation

namespace App\Message;

use Symfony\Component\Messenger\Stamp\DeduplicateStamp;

final readonly class SendWelcomeEmail
{
    public function __construct(
        public int $userId,
    ) {}
}

// 重複排除付きでディスパッチ
use Symfony\Component\Messenger\Stamp\DeduplicateStamp;

$this->bus->dispatch(
    new SendWelcomeEmail(userId: 42),
    [new DeduplicateStamp(id: 'welcome-email-42')],
);

namespace App\MessageHandler;

use App\Message\ProcessPayment;
use Symfony\Component\Messenger\Attribute\AsMessageHandler;
use Symfony\Component\Messenger\Exception\RecoverableMessageHandlingException;

#[AsMessageHandler]
final class PaymentHandler
{
    public function __invoke(ProcessPayment $message): void
    {
        try {
            $this->gateway->charge($message->amount, $message->token);
        } catch (GatewayTimeoutException $e) {
            // 回復可能：バックオフ付きでリトライ
            throw new RecoverableMessageHandlingException(
                'Payment gateway timeout, retrying',
                previous: $e,
            );
        } catch (InvalidCardException $e) {
            // 回復不可能：即座にフェイルトランスポートへ
            throw new UnrecoverableMessageHandlingException(
                'Invalid card, no retry',
                previous: $e,
            );
        }
    }
}

# 失敗したメッセージの確認と管理
php bin/console messenger:failed:show
php bin/console messenger:failed:show 20 --transport=failed

# 特定のメッセージをリトライ
php bin/console messenger:failed:retry 20 30

# クラスでフィルタして削除（Symfony 7.3+）
php bin/console messenger:failed:remove --class-filter="App\Message\CleanupTempFiles"

# 長時間処理中の再配信を防ぐためにキープアライブを有効化
php bin/console messenger:consume async --keepalive

namespace App\Message;

use Symfony\Component\Messenger\Attribute\AsMessage;

#[AsMessage(transport: 'async_priority_low')]
final readonly class GenerateReport
{
    public function __construct(
        public int $reportId,
        public string $format = 'pdf',
    ) {}
}

# config/packages/messenger.yaml
framework:
    messenger:
        transports:
            streaming:
                dsn: 'amqp-lib://guest:guest@localhost:5672/%2f/messages'
                options:
                    exchange:
                        name: app_events
                        type: topic
                    queues:
                        order_events:
                            binding_keys: ['order.*']