Основы Messenger

Компонент Messenger

Messenger -- компонент Symfony для обработки сообщений синхронно или асинхронно. Он реализует паттерн Message Bus: отправитель создаёт сообщение, шина доставляет его обработчику.

Symfony 8.0: Messenger выделен в отдельную тему экзамена. В Sf7.4 он был частью раздела Miscellaneous. Теперь это полноценная тема с транспортами, middleware и стратегией повторов.

Архитектура Messenger

Dispatch                          Handle
   |                                 |
   v                                 v
Message -> MessageBus -> Middleware -> Handler
              |                |
              v                v
         Transport         Envelope
         (async)           (stamps)

Messages (Сообщения)

Сообщение -- простой PHP-объект (DTO), содержащий данные для обработки. Не наследует интерфейсы и не содержит логики.

<?php

declare(strict_types=1);

namespace App\Message;

// Message -- simple DTO, no interface required
final readonly class SendNotification
{
    public function __construct(
        public int $userId,
        public string $subject,
        public string $content,
        public string $channel = 'email',
    ) {
    }
}

<?php

declare(strict_types=1);

namespace App\Message;

final readonly class ProcessOrder
{
    public function __construct(
        public int $orderId,
    ) {
    }
}

<?php

declare(strict_types=1);

namespace App\Message;

final readonly class GenerateReport
{
    public function __construct(
        public string $reportType,
        public \DateTimeImmutable $dateFrom,
        public \DateTimeImmutable $dateTo,
        /** @var list<string> */
        public array $filters = [],
    ) {
    }
}

Правила для сообщений

• Сообщение -- readonly DTO (только данные, без логики)
• Должно быть сериализуемым (для асинхронной обработки)
• Не содержит entity-объекты -- только ID
• Имя сообщения описывает действие: SendNotification, ProcessOrder

Подвох экзамена: Сообщение НЕ должно содержать Doctrine Entity или другие несериализуемые объекты. Передавайте только скалярные значения и ID. Обработчик сам загрузит entity из базы.

Handlers (Обработчики)

Handler -- класс, который обрабатывает конкретный тип сообщения.

Через атрибут #[AsMessageHandler]

<?php

declare(strict_types=1);

namespace App\MessageHandler;

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

#[AsMessageHandler]
final class SendNotificationHandler
{
    public function __construct(
        private readonly MailerInterface $mailer,
        private readonly UserRepository $userRepository,
    ) {
    }

    public function __invoke(SendNotification $message): void
    {
        $user = $this->userRepository->find($message->userId);

        if (null === $user) {
            return; // User not found, skip
        }

        $this->mailer->send(
            to: $user->getEmail(),
            subject: $message->subject,
            body: $message->content,
        );
    }
}

Правила привязки handler к message

Symfony определяет, какой handler обрабатывает какое сообщение по:

1. Type-hint параметра метода __invoke()
2. Атрибуту #[AsMessageHandler]

<?php

declare(strict_types=1);

namespace App\MessageHandler;

use Symfony\Component\Messenger\Attribute\AsMessageHandler;

// Handler for specific message -- determined by __invoke type-hint
#[AsMessageHandler]
final class GenerateReportHandler
{
    public function __invoke(GenerateReport $message): void
    {
        // Type-hint tells Messenger which message this handler processes
    }
}

Несколько обработчиков для одного сообщения

<?php

declare(strict_types=1);

namespace App\MessageHandler;

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

// Handler 1: Send welcome email
#[AsMessageHandler]
final class SendWelcomeEmailHandler
{
    public function __invoke(UserRegistered $message): void
    {
        // Send welcome email
    }
}

// Handler 2: Create default settings
#[AsMessageHandler]
final class CreateUserSettingsHandler
{
    public function __invoke(UserRegistered $message): void
    {
        // Create default user settings
    }
}

// Handler 3: Track analytics
#[AsMessageHandler]
final class TrackRegistrationHandler
{
    public function __invoke(UserRegistered $message): void
    {
        // Track registration event
    }
}

Message Bus

Message Bus -- центральный объект для отправки сообщений.

<?php

declare(strict_types=1);

namespace App\Controller;

use App\Message\SendNotification;
use Symfony\Component\HttpFoundation\Response;
use Symfony\Component\Messenger\MessageBusInterface;
use Symfony\Component\Routing\Attribute\Route;

final class NotificationController extends AbstractController
{
    public function __construct(
        private readonly MessageBusInterface $messageBus,
    ) {
    }

    #[Route('/notify', methods: ['POST'])]
    public function send(): Response
    {
        $this->messageBus->dispatch(
            new SendNotification(
                userId: 42,
                subject: 'Welcome!',
                content: 'Thank you for registering.',
            )
        );

        return new Response('Notification queued', Response::HTTP_ACCEPTED);
    }
}

Dispatch в сервисе

<?php

declare(strict_types=1);

namespace App\Service;

use App\Message\ProcessOrder;
use App\Message\SendNotification;
use Symfony\Component\Messenger\MessageBusInterface;

final class OrderService
{
    public function __construct(
        private readonly MessageBusInterface $messageBus,
        private readonly EntityManagerInterface $entityManager,
    ) {
    }

    public function placeOrder(Cart $cart, User $user): Order
    {
        $order = Order::fromCart($cart, $user);

        $this->entityManager->persist($order);
        $this->entityManager->flush();

        // Dispatch async messages
        $this->messageBus->dispatch(new ProcessOrder($order->getId()));

        $this->messageBus->dispatch(new SendNotification(
            userId: $user->getId(),
            subject: 'Order Confirmed',
            content: sprintf('Order #%s has been placed.', $order->getNumber()),
        ));

        return $order;
    }
}

Sync vs Async

По умолчанию сообщения обрабатываются синхронно (в том же процессе). Для асинхронной обработки нужно настроить transport.

# config/packages/messenger.yaml
framework:
    messenger:
        transports:
            async:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'

        routing:
            # Route messages to transports
            'App\Message\SendNotification': async       # Async
            'App\Message\ProcessOrder': async            # Async
            'App\Message\GenerateReport': async          # Async
            # Messages without routing are handled SYNC

Когда sync, когда async?

Envelope и Stamps

Envelope -- обёртка вокруг сообщения, содержащая метаданные (stamps).

<?php

declare(strict_types=1);

namespace App\Service;

use Symfony\Component\Messenger\Envelope;
use Symfony\Component\Messenger\MessageBusInterface;
use Symfony\Component\Messenger\Stamp\DelayStamp;
use Symfony\Component\Messenger\Stamp\TransportNamesStamp;

final class ScheduledNotificationService
{
    public function __construct(
        private readonly MessageBusInterface $bus,
    ) {
    }

    public function scheduleNotification(
        SendNotification $message,
        int $delayMs = 0,
    ): void {
        $envelope = new Envelope($message, [
            // Delay processing by N milliseconds
            new DelayStamp($delayMs),
        ]);

        $this->bus->dispatch($envelope);
    }

    public function sendToSpecificTransport(SendNotification $message): void
    {
        $envelope = new Envelope($message, [
            // Force a specific transport
            new TransportNamesStamp(['high_priority']),
        ]);

        $this->bus->dispatch($envelope);
    }
}

Основные Stamps

Именованные шины

framework:
    messenger:
        default_bus: command.bus

        buses:
            command.bus:
                middleware:
                    - doctrine_transaction

            query.bus:
                middleware: []

            event.bus:
                default_middleware:
                    allow_no_handlers: true

<?php

declare(strict_types=1);

namespace App\Controller;

use Symfony\Component\Messenger\MessageBusInterface;
use Symfony\Component\DependencyInjection\Attribute\Autowire;

final class ProductController extends AbstractController
{
    public function __construct(
        #[Autowire(service: 'command.bus')]
        private readonly MessageBusInterface $commandBus,
        #[Autowire(service: 'query.bus')]
        private readonly MessageBusInterface $queryBus,
    ) {
    }
}

Итоги

• Message -- readonly DTO с данными (без entity, без логики)
• Handler -- __invoke(Message) с атрибутом #[AsMessageHandler]
• MessageBus -- dispatch() отправляет сообщение обработчику
• Sync по умолчанию, async через transport routing
• Envelope содержит stamps (метаданные): delay, transport и др.
• Один message может иметь несколько handlers