Messenger · Symfony

Messenger надає шину повідомлень (message bus) із можливістю надсилати повідомлення й потім обробляти їх одразу у вашому застосунку або передавати їх через транспорти (наприклад, черги), щоб обробити пізніше. Компонент значною мірою натхненний серією блогових дописів про командні шини (blog posts about command buses) Матіаса Нобака та проєктом SimpleBus.

Встановлення

У застосунках, що використовують Symfony Flex, виконайте цю команду, щоб встановити messenger:

composer require symfony/messenger

Концепції

Перед використанням Messenger корисно ознайомитися з його основними концепціями та тим, як вони повʼязані між собою:

Sender (відправник): : Відповідає за серіалізацію та надсилання повідомлень кудись. Цим «кудись» може бути, наприклад, брокер повідомлень або стороннє API.

Receiver (отримувач): : Відповідає за отримання, десеріалізацію та передавання повідомлень до обробника (обробників). Це може бути, наприклад, споживач черги повідомлень або точка входу API.

Handler (обробник): : Відповідає за обробку повідомлень із застосуванням бізнес-логіки, що стосується цих повідомлень. Обробники викликаються middleware HandleMessageMiddleware.

Middleware: : Middleware має доступ до повідомлення та його обгортки (конверта, envelope) під час його проходження через шину. Буквально «програмне забезпечення посередині» — воно не стосується основних завдань (бізнес-логіки) застосунку. Натомість це наскрізні аспекти, застосовні в межах усього застосунку і такі, що впливають на всю шину повідомлень. Наприклад: логування, валідація повідомлення, старт транзакції, ... Вони також відповідають за виклик наступного middleware в ланцюжку, а отже можуть змінювати конверт, додаючи до нього марки (stamps) або навіть замінюючи його, а також переривати ланцюжок middleware. Middleware викликається як тоді, коли повідомлення початково відправляється, так і пізніше, коли повідомлення отримано з транспорту.

Envelope (конверт): : Специфічна для Messenger концепція, що дає повну гнучкість усередині шини повідомлень, обгортаючи повідомлення й дозволяючи додавати всередину корисну інформацію через марки конверта (envelope stamps).

Envelope Stamps (марки конверта): : Частинка інформації, яку потрібно прикріпити до повідомлення: контекст серіалізатора для транспорту, маркери, що ідентифікують отримане повідомлення, або будь-які метадані, які може використовувати ваш middleware чи транспортний рівень.

Створення повідомлення та обробника

Messenger побудований довкола двох різних класів, які ви створюєте: (1) класу повідомлення, що містить дані, і (2) класу обробника (обробників), який буде викликаний, коли це повідомлення буде відправлено. Клас обробника читає клас повідомлення і виконує одне або кілька завдань.

До класу повідомлення немає особливих вимог, окрім того, що він має бути серіалізовним:

// src/Message/SmsNotification.php
namespace App\Message;

class SmsNotification
{
    public function __construct(
        private string $content,
    ) {
    }

    public function getContent(): string
    {
        return $this->content;
    }
}

Обробник повідомлення — це PHP callable; рекомендований спосіб його створення — створити клас, що має атрибут Symfony\Component\Messenger\Attribute\AsMessageHandler і метод __invoke() із тип-хінтом класу повідомлення (або інтерфейсу повідомлення):

// src/MessageHandler/SmsNotificationHandler.php
namespace App\MessageHandler;

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

#[AsMessageHandler]
class SmsNotificationHandler
{
    public function __invoke(SmsNotification $message)
    {
        // ... виконати якусь роботу — наприклад, надіслати SMS-повідомлення!
    }
}

Порада. Ви також можете використовувати атрибут #[AsMessageHandler] на окремих методах класу. Ви можете застосувати атрибут до будь-якої кількості методів в одному класі, що дозволяє згрупувати обробку кількох повʼязаних типів повідомлень.

Завдяки автоконфігурації та тип-хінту SmsNotification Symfony знає, що цей обробник має бути викликаний, коли відправляється повідомлення SmsNotification. У більшості випадків це все, що потрібно зробити. Але ви також можете налаштувати обробники повідомлень вручну. Щоб побачити всі налаштовані обробники, виконайте:

php bin/console debug:messenger

Відправлення повідомлення

Ви готові! Щоб відправити повідомлення (і викликати обробник), впровадьте сервіс messenger.default_bus (через MessageBusInterface), наприклад у контролері. Коли ви використовуєте Messenger у довільному PHP-застосунку, створіть шину повідомлень самостійно і зареєструйте в ній обробник(и) (у такому разі атрибут #[AsMessageHandler] не потрібен):

// src/Controller/DefaultController.php
namespace App\Controller;

use App\Message\SmsNotification;
use Symfony\Bundle\FrameworkBundle\Controller\AbstractController;
use Symfony\Component\HttpFoundation\Response;
use Symfony\Component\Messenger\MessageBusInterface;

class DefaultController extends AbstractController
{
    public function index(MessageBusInterface $bus): Response
    {
        // призведе до виклику SmsNotificationHandler
        $bus->dispatch(new SmsNotification('Look! I created a message!'));

        // ...
    }
}

use App\Message\SmsNotification;
use App\MessageHandler\SmsNotificationHandler;
use Symfony\Component\Messenger\Handler\HandlersLocator;
use Symfony\Component\Messenger\MessageBus;
use Symfony\Component\Messenger\Middleware\HandleMessageMiddleware;

$bus = new MessageBus([
    new HandleMessageMiddleware(new HandlersLocator([
        SmsNotification::class => [new SmsNotificationHandler()],
    ])),
]);

// призведе до виклику SmsNotificationHandler
$bus->dispatch(new SmsNotification('Look! I created a message!'));

Транспорти: асинхронні повідомлення та повідомлення в черзі

За замовчуванням повідомлення обробляються одразу після відправлення. Якщо ви хочете обробляти повідомлення асинхронно, можете налаштувати транспорт. Транспорт здатний надсилати повідомлення (наприклад, до системи черг), а потім отримувати їх через обробник черги (worker). Messenger підтримує кілька транспортів.

Примітка. Якщо ви хочете використати транспорт, який не підтримується, погляньте на транспорт Enqueue (Enqueue's transport), що підтримує такі сервіси, як Kafka та Google Pub/Sub.

Транспорт реєструється за допомогою «DSN». Завдяки рецепту Flex для Messenger у вашому файлі .env уже є кілька прикладів.

# MESSENGER_TRANSPORT_DSN=amqp://guest:guest@localhost:5672/%2f/messages
# MESSENGER_TRANSPORT_DSN=doctrine://default
# MESSENGER_TRANSPORT_DSN=redis://localhost:6379/messages

Розкоментуйте той транспорт, який вам потрібен (або задайте його в .env.local). Докладніше див. розділ про конфігурацію транспортів.

Далі, у config/packages/messenger.yaml, визначмо транспорт з назвою async, який використовує цю конфігурацію:

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

            # або в розгорнутому вигляді, щоб налаштувати більше опцій
            #async:
            #    dsn: "%env(MESSENGER_TRANSPORT_DSN)%"
            #    options: []

// config/packages/messenger.php
namespace Symfony\Component\DependencyInjection\Loader\Configurator;

return App::config([
    'framework' => [
        'messenger' => [
            'transports' => [
                'async' => env('MESSENGER_TRANSPORT_DSN'),

                // або в розгорнутому вигляді, щоб налаштувати більше опцій
                // 'async' => [
                //    'dsn' => env('MESSENGER_TRANSPORT_DSN'),
                //    'options' => [],
                // ],
            ],
        ],
    ],
]);

Маршрутизація повідомлень до транспорту

Тепер, коли транспорт налаштовано, замість негайної обробки повідомлення ви можете налаштувати їх надсилання до транспорту:

// src/Message/SmsNotification.php
namespace App\Message;

use Symfony\Component\Messenger\Attribute\AsMessage;

#[AsMessage('async')]
class SmsNotification
{
    // ...
}

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

        routing:
            # async — це будь-яка назва, яку ви дали своєму транспорту вище
            'App\Message\SmsNotification': async

// config/packages/messenger.php
namespace Symfony\Component\DependencyInjection\Loader\Configurator;

use App\Message\SmsNotification;

return App::config([
    'framework' => [
        'messenger' => [
            'routing' => [
                // async — це будь-яка назва, яку ви дали своєму транспорту вище
                SmsNotification::class => 'async',
            ],
        ],
    ],
]);

Завдяки цьому App\Message\SmsNotification буде надіслано до транспорту async, і його обробник(и) не буде викликано одразу. Будь-які повідомлення, що не збігаються з правилами в routing, і надалі оброблятимуться негайно, тобто синхронно.

Примітка. Якщо ви налаштовуєте маршрутизацію одночасно у файлах конфігурації YAML/PHP і через PHP-атрибути, конфігурація завжди має пріоритет над атрибутом класу. Така поведінка дозволяє перевизначати маршрутизацію окремо для кожного середовища.

Примітка. Налаштовуючи маршрутизацію в окремих файлах YAML/PHP, ви можете використати частковий простір імен PHP, як-от 'App\Message\*', щоб охопити всі повідомлення у відповідному просторі імен. Єдина вимога — символ підстановки '*' має стояти в кінці простору імен.

Ви можете використати '*' як клас повідомлення. Це працюватиме як типове правило маршрутизації для будь-якого повідомлення, що не збіглося з правилами в routing. Це корисно, щоб гарантувати, що жодне повідомлення не оброблятиметься синхронно за замовчуванням.

Єдиний недолік у тому, що '*' застосується також до листів, надісланих Symfony Mailer (який використовує SendEmailMessage, коли Messenger доступний). Це може спричинити проблеми, якщо ваші листи не серіалізовні (наприклад, якщо вони містять вкладення у вигляді PHP-ресурсів/потоків).

Ви також можете маршрутизувати класи за їхнім батьківським класом або інтерфейсом. Або надсилати повідомлення до кількох транспортів:

// src/Message/SmsNotification.php
namespace App\Message;

use Symfony\Component\Messenger\Attribute\AsMessage;

#[AsMessage(['async', 'audit'])]
class SmsNotification
{
    // ...
}

// якщо вам так зручніше, ви також можете застосувати кілька атрибутів до класу повідомлення
#[AsMessage('async')]
#[AsMessage('audit')]
class SmsNotification
{
    // ...
}

# config/packages/messenger.yaml
framework:
    messenger:
        routing:
            # маршрутизувати всі повідомлення, що успадковують цей приклад базового класу або інтерфейсу
            'App\Message\AbstractAsyncMessage': async
            'App\Message\AsyncMessageInterface': async

            'My\Message\ToBeSentToTwoSenders': [async, audit]

// config/packages/messenger.php
namespace Symfony\Component\DependencyInjection\Loader\Configurator;

use App\Message\AbstractAsyncMessage;
use App\Message\AsyncMessageInterface;
use My\Message\ToBeSentToTwoSenders;

return App::config([
    'framework' => [
        'messenger' => [
            'routing' => [
                // маршрутизувати всі повідомлення, що успадковують цей приклад базового класу або інтерфейсу
                AbstractAsyncMessage::class => 'async',
                AsyncMessageInterface::class => 'async',
                ToBeSentToTwoSenders::class => ['async', 'audit'],
            ],
        ],
    ],
]);

Примітка. Якщо ви налаштуєте маршрутизацію і для дочірнього, і для батьківського класу, буде використано обидва правила. Наприклад, якщо у вас є обʼєкт SmsNotification, що успадковує Notification, буде використано маршрутизацію і для Notification, і для SmsNotification.

Порада. Ви можете визначати й перевизначати транспорт, який використовує повідомлення, під час виконання за допомогою Symfony\Component\Messenger\Stamp\TransportNamesStamp на конверті повідомлення. Ця марка приймає масив назв транспортів як єдиний аргумент. Докладніше про марки див. «Конверти та марки» (Envelopes & Stamps).

Сутності Doctrine у повідомленнях

Якщо вам потрібно передати сутність Doctrine у повідомленні, краще передати первинний ключ сутності (або будь-яку релевантну інформацію, яка справді потрібна обробнику, наприклад email тощо) замість обʼєкта (інакше ви можете побачити помилки, повʼязані з Entity Manager):

// src/Message/NewUserWelcomeEmail.php
namespace App\Message;

class NewUserWelcomeEmail
{
    public function __construct(
        private int $userId,
    ) {
    }

    public function getUserId(): int
    {
        return $this->userId;
    }
}

Потім у своєму обробнику ви можете запитати свіжий обʼєкт:

// src/MessageHandler/NewUserWelcomeEmailHandler.php
namespace App\MessageHandler;

use App\Message\NewUserWelcomeEmail;
use App\Repository\UserRepository;
use Symfony\Component\Messenger\Attribute\AsMessageHandler;

#[AsMessageHandler]
class NewUserWelcomeEmailHandler
{
    public function __construct(
        private UserRepository $userRepository,
    ) {
    }

    public function __invoke(NewUserWelcomeEmail $welcomeEmail): void
    {
        $user = $this->userRepository->find($welcomeEmail->getUserId());

        // ... надіслати лист!
    }
}

Це гарантує, що сутність містить свіжі дані.

Версіонування класів повідомлень

Клас повідомлення визначає контракт між кодом, який відправляє повідомлення, і обробником черги, який його обробляє. Оскільки Messenger обробляє повідомлення асинхронно, деякі повідомлення можуть усе ще очікувати в черзі, коли ви розгортаєте нову версію застосунку. Якщо ви зміните клас повідомлення, ті старіші повідомлення можуть більше не десеріалізуватися коректно, що може призвести до збоїв, коли обробники черги спробують їх обробити.

Для незначних змін зберігайте зворотну сумісність, роблячи нові аргументи конструктора необовʼязковими та надаючи розумне значення за замовчуванням:

final class SendInvoice
{
    public function __construct(
        public readonly int $orderId,
        public readonly ?string $locale = null, // додано пізніше
    ) {
    }
}

За такого підходу старіші повідомлення, що не містять нового аргументу $locale, і надалі можуть коректно десеріалізуватися.

Ще одна зміна, що ламає сумісність, — це видалення властивостей із класу повідомлення. Повідомлення, вже збережені в черзі, можуть усе ще містити ці властивості, коли їх десеріалізують новіші обробники черги. Починаючи з PHP 8.2, це може спричинити попередження про застарілість, оскільки динамічні властивості є застарілими.

Якщо вам необхідно видалити властивість, розгляньте один із таких підходів:

• Тимчасово залишити властивість (наприклад, як public ?type $property = null), доки всі старі повідомлення не буде оброблено.
• Додати до класу атрибут #[\AllowDynamicProperties], щоб дозволити старішим серіалізованим повідомленням задавати властивості, яких більше не існує.
• Реалізувати власну логіку серіалізації, щоб контролювати, як повідомлення серіалізується й десеріалізується.

Якщо зміна змінює сенс повідомлення, а не просто розширює його, створіть нову версію класу повідомлення замість того, щоб змінювати наявний:

// Зберігайте SendInvoice, доки не буде оброблено всі повідомлення в черзі, що його використовують
final class SendInvoiceV2
{
    public function __construct(
        public readonly int $orderId,
        public readonly string $locale,
        public readonly string $templateId,
    ) {
    }
}

Під час розгортання обидві версії можуть тимчасово співіснувати. Спочатку розгорніть новий клас повідомлення SendInvoiceV2 та його обробник, зберігаючи старі. Після того як чергу буде повністю спорожнено, видаліть старий клас і обробник у наступному розгортанні.

Занадто раннє видалення старого класу може призвести до збоїв обробників черги, коли вони намагатимуться десеріалізувати повідомлення, відправлені до розгортання.

Синхронна обробка повідомлень

Якщо повідомлення не збігається з жодним правилом маршрутизації, воно не буде надіслане до жодного транспорту і буде оброблене негайно. У деяких випадках (наприклад, коли привʼязуєте обробники до різних транспортів) простіше або гнучкіше зробити це явно: створити транспорт sync і «надсилати» повідомлення туди, щоб вони оброблялися негайно:

# config/packages/messenger.yaml
framework:
    messenger:
        transports:
            # ... інші транспорти

            sync: 'sync://'

        routing:
            App\Message\SmsNotification: sync

// config/packages/messenger.php
namespace Symfony\Component\DependencyInjection\Loader\Configurator;

use App\Message\SmsNotification;

return App::config([
    'framework' => [
        'messenger' => [
            'transports' => [
                'sync' => 'sync://',
            ],
            'routing' => [
                SmsNotification::class => 'sync',
            ],
        ],
    ],
]);

Створення власного транспорту

Ви також можете створити власний транспорт, якщо вам потрібно надсилати чи отримувати повідомлення від чогось, що не підтримується. Див. /messenger/custom-transport.

Споживання повідомлень (запуск обробника черги)

Щойно ваші повідомлення промаршрутизовано, у більшості випадків вам потрібно буде їх «спожити». Це можна зробити командою messenger:consume:

php bin/console messenger:consume async

# використайте -vv, щоб побачити подробиці того, що відбувається
php bin/console messenger:consume async -vv

# використайте регулярний вираз, щоб охопити кілька транспортів одразу
# (наприклад, усі транспорти, що починаються з "scheduler_")
php bin/console messenger:consume scheduler_.*

# поєднуйте регулярні вирази та явні назви транспортів
php bin/console messenger:consume high_priority.* low_priority async

Перший аргумент — це назва отримувача (або id сервісу, якщо ви маршрутизували до власного сервісу). За замовчуванням команда працюватиме вічно: шукатиме нові повідомлення у вашому транспорті й оброблятиме їх. Ця команда й називається вашим «обробником черги» (worker).

Ви також можете використовувати регулярні вирази як назву отримувача, щоб охопити кілька транспортів одразу. Це корисно, коли у вас є кілька транспортів зі схожими назвами, наприклад транспорти, згруповані за призначенням або пріоритетом.

Коли регулярний вираз відповідає кільком транспортам, вони споживаються в тому порядку, у якому визначені у вашій конфігурації. Якщо ви вкажете кілька назв отримувачів або регулярних виразів, вони обробляються в тому порядку, у якому ви передаєте їх команді.

Примітка. Зіставлення з регулярними виразами працює лише тоді, коли команда не запущена в інтерактивному режимі.

Нове у версії 8.1. Підтримку регулярних виразів як назви отримувача було додано в Symfony 8.1.

Якщо ви хочете споживати повідомлення з усіх доступних отримувачів, можете використати команду з опцією --all:

php bin/console messenger:consume --all

Використовуючи --all, ви можете виключити окремих отримувачів за допомогою опції --exclude-receivers (скорочення -eq):

php bin/console messenger:consume --all --exclude-receivers=async_priority_low --exclude-receivers=failed

Примітка. Опцію --exclude-receivers можна використовувати лише разом із --all. Крім того, ви не можете виключити всіх отримувачів.

Повідомлення, обробка яких триває довго, можуть бути передчасно доставлені повторно, оскільки деякі транспорти вважають, що непідтверджене повідомлення втрачене. Щоб запобігти цій проблемі, використовуйте опцію команди --keepalive, щоб задати інтервал (у секундах; значення за замовчуванням = 5), з яким повідомлення позначається як «у процесі». Це запобігає повторній доставці повідомлення, доки обробник черги не завершить його обробку:

php bin/console messenger:consume --keepalive

Примітка. Ця опція доступна лише для таких транспортів: Beanstalkd, AmazonSQS, Doctrine і Redis.

За замовчуванням обробник черги забирає з транспорту одне повідомлення за ітерацію. Використовуйте опцію --fetch-size, щоб забирати кілька повідомлень за ітерацію, зменшуючи кількість звернень до транспорту:

php bin/console messenger:consume async --fetch-size=8

Це особливо корисно з транспортами, які підтримують отримання кількох повідомлень за один виклик, як-от Amazon SQS, Redis, AMQP і Doctrine.

Нове у версії 8.1. Опцію --fetch-size було додано в Symfony 8.1.

Порада. У середовищі розробки, якщо ви користуєтеся інструментом Symfony CLI, ви можете налаштувати автоматичний запуск обробників черги разом із вебсервером. Докладніше — у документації «Symfony CLI Workers».

Порада. Щоб коректно зупинити обробник черги, викиньте екземпляр Symfony\Component\Messenger\Exception\StopWorkerException.

Розгортання у продакшн

У продакшні є кілька важливих речей, про які варто подумати:

Використовуйте менеджер процесів, як-от Supervisor або systemd, щоб ваші обробники черги працювали постійно

Вам потрібно, щоб один чи більше «обробників черги» працювали весь час. Для цього використовуйте систему контролю процесів, як-от Supervisor або systemd.

Не дозволяйте обробникам черги працювати вічно

Деякі сервіси (як-от EntityManager від Doctrine) з часом споживають дедалі більше памʼяті. Тож замість того, щоб дозволяти обробнику черги працювати вічно, використайте прапорець на кшталт messenger:consume --limit=10, щоб сказати обробнику обробити лише 10 повідомлень перед виходом (після чого менеджер процесів створить новий процес). Є також інші опції, як-от --memory-limit=128M і --time-limit=3600.

Зупинка обробників черги, що натрапляють на помилки

Якщо залежність обробника черги, наприклад ваш сервер бази даних, недоступна або досягнуто таймауту, ви можете спробувати додати логіку повторного підключення або просто завершувати обробник, якщо він отримує забагато помилок, — за допомогою опції --failure-limit команди messenger:consume.

Перезапускайте обробники черги під час розгортання

Щоразу під час розгортання перезапускайте всі процеси обробників черги, щоб вони підхопили новий код. Для цього виконайте messenger:stop-workers після того, як новий код опиниться на диску й кеш буде прогрітий, але до перемикання трафіку. Це встановлює прапорець у кеші, який каже кожному обробнику завершити поточне повідомлення й коректно вийти. Далі менеджер процесів перезапускає їх уже на новій кодовій базі:

# у вашому скрипті розгортання, після викладення коду й прогріву кешу:
php bin/console messenger:stop-workers

Команда внутрішньо використовує кеш застосунку (app cache). Якщо ваш застосунок працює на кількох хостах, налаштуйте кеш застосунку на використання спільного адаптера (наприклад, Redis), щоб усі вебпроцеси та процеси обробників черги використовували один і той самий кеш.

Примітка. У середовищі Kubernetes послідовний перезапуск (rolling restart) Deployment обробників черги дає той самий результат, але лише якщо terminationGracePeriodSeconds достатньо великий, щоб найдовший обробник встиг завершитися до заміни пода. SIGKILL не дає обробникам черги шансу завершити поточне повідомлення, через що обробник може лишитися посеред виконання.

Використовуйте той самий кеш між розгортаннями

Якщо ваша стратегія розгортання передбачає створення нових цільових директорій, вам слід задати значення для опції конфігурації cache.prefix_seed, щоб використовувати той самий простір імен кешу між розгортаннями. Інакше пул cache.app використовуватиме значення параметра kernel.project_dir як основу для простору імен, що призводитиме до різних просторів імен щоразу під час нового розгортання.

Пріоритезовані транспорти

Використовуйте окремі транспорти для типів повідомлень із різними вимогами до затримки, різними режимами відмов чи вікнами повторних спроб. Коли кілька типів повідомлень поділяють один транспорт, повільний чи проблемний обробник одного типу може затримувати всі інші в тій самій черзі.

Наприклад, якщо повідомлення синхронізації каталогу й підтвердження платежів маршрутизуються до одного транспорту, повільний обробник товарного фіду може затримати підтвердження платежів для клієнтів, які проходять оформлення замовлення.

Розгляньте призначення кожному транспорту власного обробника черги, щоб збої чи сповільнення в одному потоці повідомлень не впливали на інші:

# config/packages/messenger.yaml
framework:
    messenger:
        transports:
            async_priority_high:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                options:
                    # queue_name є специфічним для транспорту doctrine
                    queue_name: high

                    # для AMQP надсилайте до окремого exchange, потім до черги
                    #exchange:
                    #    name: high
                    #queues:
                    #    messages_high: ~
                    # для redis спробуйте "group"
            async_priority_low:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                options:
                    queue_name: low

        routing:
            'App\Message\SmsNotification': async_priority_low
            'App\Message\NewUserWelcomeEmail': async_priority_high

// config/packages/messenger.php
namespace Symfony\Component\DependencyInjection\Loader\Configurator;

use App\Message\NewUserWelcomeEmail;
use App\Message\SmsNotification;

return App::config([
    'framework' => [
        'messenger' => [
            'transports' => [
                'async_priority_high' => [
                    'dsn' => env('MESSENGER_TRANSPORT_DSN'),
                    'options' => [
                        // queue_name є специфічним для транспорту doctrine
                        'queue_name' => 'high',

                        // для AMQP надсилайте до окремого exchange, потім до черги
                        // 'exchange' => [
                        //    'name' => 'high',
                        // ],
                        // 'queues' => [
                        //    'messages_high' => null,
                        // ],
                        // для redis спробуйте "group"
                    ],
                ],
                'async_priority_low' => [
                    'dsn' => env('MESSENGER_TRANSPORT_DSN'),
                    'options' => [
                        'queue_name' => 'low',
                    ],
                ],
            ],
            'routing' => [
                SmsNotification::class => 'async_priority_low',
                NewUserWelcomeEmail::class => 'async_priority_high',
            ],
        ],
    ],
]);

Далі ви можете запускати окремі обробники черги для кожного транспорту або вказати одному обробнику обробляти повідомлення в порядку пріоритету:

php bin/console messenger:consume async_priority_high async_priority_low

Обробник черги завжди спершу шукатиме повідомлення, що очікують в async_priority_high. Якщо їх немає, тоді він споживатиме повідомлення з async_priority_low.

Пріоритезовані повідомлення

Нове у версії 8.1. Підтримку пріоритезації повідомлень було додано в Symfony 8.1.

За замовчуванням Messenger використовує порядок «перший прийшов — перший вийшов», тож повідомлення отримуються в тому самому порядку, у якому були надіслані, за винятком відкладених повідомлень.

Пріоритезовані транспорти дають базову підтримку пріоритетів, але лише між різними типами повідомлень. Використовуючи транспорти AMQP або Beanstalkd, ви можете пріоритезувати повідомлення одного типу за допомогою марки під назвою PriorityStamp:

use Symfony\Component\Messenger\Envelope;
use Symfony\Component\Messenger\Stamp\PriorityStamp;

$bus->dispatch(
    (new Envelope($message))->with(new PriorityStamp(255))
);

Порада. Підтримувані пріоритети — у діапазоні від 0 (найнижчий) до 255 (найвищий).

PriorityStamp можна поєднувати з DelayStamp. Коли затримка спливає, повідомлення доставляється раніше за будь-які повідомлення з нижчим пріоритетом, що вже перебувають у черзі:

use Symfony\Component\Messenger\Envelope;
use Symfony\Component\Messenger\Stamp\DelayStamp;
use Symfony\Component\Messenger\Stamp\PriorityStamp;

$bus->dispatch(
    (new Envelope($message))->with(
        new PriorityStamp(255),
        new DelayStamp(5000)
    )
);

Під час використання транспорту Beanstalkd додаткова конфігурація не потрібна. Beanstalkd внутрішньо використовує обернену шкалу пріоритетів (0 = найвищий, 2^32 - 1 = найнижчий), але Messenger виконує перетворення автоматично.

Під час використання транспорту AMQP черги з пріоритетами мають бути явно увімкнені в конфігурації:

# config/packages/messenger.yaml
framework:
    messenger:
        transports:
            async:
                dsn: "%env(MESSENGER_TRANSPORT_DSN)%"
                options:
                    queues:
                        messenger:
                            arguments:
                                x-max-priority: 255

// config/packages/messenger.php
use Symfony\Config\FrameworkConfig;

return static function (FrameworkConfig $framework): void {
    $framework->messenger()
        ->transport('async')
        ->dsn('%env(MESSENGER_TRANSPORT_DSN)%')
        ->options([
            'queues' => [
                'messenger' => [
                    'arguments' => ['x-max-priority' => 255],
                ],
            ],
        ]);
};

Увага! x-max-priority не можна змінити на наявній черзі RabbitMQ. Messenger не зможе виконати автоматичне налаштування, і пріоритети не працюватимуть. Натомість створіть нову чергу.

Примітка. RabbitMQ рекомендує використовувати не більше 10 рівнів пріоритету (no more than 10 priority levels). Наприклад: 255 для високого, 127 для середнього і 0 для низького. Більша кількість рівнів може вплинути на продуктивність.

Обмеження споживання конкретними чергами

Деякі транспорти (зокрема AMQP) мають поняття exchange і черг. Транспорт Symfony завжди привʼязаний до exchange. За замовчуванням обробник черги споживає з усіх черг, приєднаних до exchange вказаного транспорту. Проте бувають випадки, коли потрібно, щоб обробник споживав лише з певних черг.

Ви можете обмежити обробник черги, щоб він обробляв повідомлення лише з конкретної черги (черг):

php bin/console messenger:consume my_transport --queues=fasttrack

# ви можете передати опцію --queues більш ніж один раз, щоб обробляти кілька черг
php bin/console messenger:consume my_transport --queues=fasttrack1 --queues=fasttrack2

Примітка. Щоб можна було використовувати опцію queues, отримувач має реалізовувати Symfony\Component\Messenger\Transport\Receiver\QueueReceiverInterface.

Перевірка кількості повідомлень у черзі для кожного транспорту

Виконайте команду messenger:stats, щоб дізнатися, скільки повідомлень перебуває в «чергах» деяких або всіх транспортів:

# показує кількість повідомлень у черзі в усіх транспортах
php bin/console messenger:stats

# показує статистику лише для деяких транспортів
php bin/console messenger:stats my_transport_name other_transport_name

# ви також можете вивести статистику у форматі JSON
php bin/console messenger:stats --format=json
php bin/console messenger:stats my_transport_name other_transport_name --format=json

Примітка. Щоб ця команда працювала, отримувач налаштованого транспорту має реалізовувати Symfony\Component\Messenger\Transport\Receiver\MessageCountAwareInterface.

Конфігурація Supervisor

Supervisor — чудовий інструмент, щоб гарантувати, що ваш процес (процеси) обробника черги завжди працює (навіть якщо він завершується через збій, досягнення ліміту повідомлень або завдяки messenger:stop-workers). Ви можете встановити його, наприклад, в Ubuntu через:

sudo apt-get install supervisor

Файли конфігурації Supervisor зазвичай розташовані в директорії /etc/supervisor/conf.d. Наприклад, ви можете створити там новий файл messenger-worker.conf, щоб переконатися, що 2 екземпляри messenger:consume працюють увесь час:

;/etc/supervisor/conf.d/messenger-worker.conf
[program:messenger-consume]
command=php /path/to/your/app/bin/console messenger:consume async --time-limit=3600
user=ubuntu
numprocs=2
startsecs=0
autostart=true
autorestart=true
startretries=10
process_name=%(program_name)s_%(process_num)02d

Змініть аргумент async на назву вашого транспорту (або транспортів), а user — на Unix-користувача на вашому сервері.

Увага! Під час розгортання щось може бути недоступним (наприклад, база даних), через що споживач не зможе запуститися. У такій ситуації Supervisor спробує перезапустити команду startretries разів. Обовʼязково змініть цю настройку, щоб команда не потрапила у стан FATAL, з якого вона більше ніколи не перезапуститься.

З кожним перезапуском Supervisor збільшує затримку на 1 секунду. Наприклад, якщо значення дорівнює 10, він чекатиме 1 с, 2 с, 3 с тощо. Це дає сервісу загалом 55 секунд, щоб знову стати доступним. Збільште настройку startretries, щоб покрити максимально очікуваний простій.

Якщо ви використовуєте транспорт Redis, зауважте, що кожному обробнику черги потрібне унікальне імʼя споживача, щоб уникнути обробки одного й того самого повідомлення кількома обробниками. Один зі способів цього досягти — задати змінну середовища у файлі конфігурації Supervisor, на яку потім можна посилатися в messenger.yaml (див. розділ про Redis нижче):

environment=MESSENGER_CONSUMER_NAME=%(program_name)s_%(process_num)02d

Далі скажіть Supervisor прочитати вашу конфігурацію і запустити ваші обробники черги:

sudo supervisorctl reread

sudo supervisorctl update

sudo supervisorctl start messenger-consume:*

# Якщо ви розгортаєте оновлення свого коду, не забудьте перезапустити обробники черги,
# щоб запустити новий код
sudo supervisorctl restart messenger-consume:*

Докладніше — у документації Supervisor (Supervisor docs).

Коректне завершення роботи

Якщо ви встановите у своєму проєкті PHP-розширення PCNTL, обробники черги оброблятимуть POSIX-сигнали SIGTERM або SIGINT, щоб завершити обробку поточного повідомлення перед припиненням роботи.

Проте ви можете віддавати перевагу іншим POSIX-сигналам для коректного завершення. Ви можете перевизначити типові, задавши опцію конфігурації framework.messenger.stop_worker_on_signals:

# config/packages/messenger.yaml
framework:
    messenger:
        stop_worker_on_signals:
            - SIGTERM
            - SIGINT
            - SIGUSR1

    # ...

<!-- config/packages/messenger.xml -->
<?xml version="1.0" encoding="UTF-8" ?>
<container xmlns="http://symfony.com/schema/dic/services"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:framework="http://symfony.com/schema/dic/symfony"
    xsi:schemaLocation="http://symfony.com/schema/dic/services
        https://symfony.com/schema/dic/services/services-1.0.xsd
        http://symfony.com/schema/dic/symfony
        https://symfony.com/schema/dic/symfony/symfony-1.0.xsd">

    <framework:config>
        <framework:messenger>
            <!-- ... -->

            <framework:stop-worker-on-signals signal="SIGTERM"/>
            <framework:stop-worker-on-signals signal="SIGINT"/>
            <framework:stop-worker-on-signals signal="SIGUSR1"/>
        </framework:messenger>
    </framework:config>
</container>

// config/packages/messenger.php
namespace Symfony\Component\DependencyInjection\Loader\Configurator;

return App::config([
    'framework' => [
        'messenger' => [
            'stop_worker_on_signals' => [
                'SIGTERM',
                'SIGINT',
                'SIGUSR1',
            ],
        ],
    ],
]);

У деяких випадках сигнал SIGTERM надсилає сам Supervisor (наприклад, під час зупинки Docker-контейнера, у якого Supervisor є entrypoint). У таких випадках вам потрібно додати ключ stopwaitsecs до конфігурації програми (зі значенням бажаного пільгового періоду в секундах), щоб виконати коректне завершення роботи:

[program:x]
stopwaitsecs=20

Конфігурація systemd

Хоча Supervisor — чудовий інструмент, він має недолік: щоб його запустити, потрібен доступ до системи. Systemd став стандартом у більшості дистрибутивів Linux і має хорошу альтернативу — користувацькі сервіси (user services).

Файли конфігурації користувацьких сервісів systemd зазвичай розташовані в директорії ~/.config/systemd/user. Наприклад, ви можете створити новий файл messenger-worker.service. Або файл messenger-worker@.service, якщо хочете, щоб одночасно працювало більше екземплярів:

[Unit]
Description=Symfony messenger-consume %i

[Service]
ExecStart=php /path/to/your/app/bin/console messenger:consume async --time-limit=3600
# для Redis задайте власне імʼя споживача для кожного екземпляра
Environment="MESSENGER_CONSUMER_NAME=symfony-%n-%i"
Restart=always
RestartSec=30

[Install]
WantedBy=default.target

Тепер скажіть systemd увімкнути й запустити один обробник черги:

systemctl --user enable messenger-worker@1.service
systemctl --user start messenger-worker@1.service

# щоб увімкнути й запустити 20 обробників черги
systemctl --user enable messenger-worker@{1..20}.service
systemctl --user start messenger-worker@{1..20}.service

Якщо ви зміните файл конфігурації сервісу, вам потрібно перезавантажити демон:

systemctl --user daemon-reload

Щоб перезапустити всіх ваших споживачів:

systemctl --user restart messenger-consume@*.service

Користувацький екземпляр systemd запускається лише після першого входу відповідного користувача в систему. Натомість споживачам часто потрібно запускатися під час завантаження системи. Увімкніть lingering для користувача, щоб активувати таку поведінку:

loginctl enable-linger <your-username>

Логами керує journald, і з ними можна працювати за допомогою команди journalctl:

# стежити за логами споживача nr 11
journalctl -f --user-unit messenger-consume@11.service

# стежити за логами всіх споживачів
journalctl -f --user-unit messenger-consume@*

# стежити за всіма логами ваших користувацьких сервісів
journalctl -f _UID=$UID

Докладніше — у документації systemd (systemd docs).

Примітка. Вам потрібні або підвищені привілеї для команди journalctl, або додайте свого користувача до групи systemd-journal:

sudo usermod -a -G systemd-journal <your-username>

Обробник черги без стану

PHP спроєктовано як мову без стану; між різними запитами немає спільних ресурсів. У HTTP-контексті PHP очищає все після надсилання відповіді, тож ви можете вирішити не перейматися сервісами, які можуть спричиняти витік памʼяті.

З іншого боку, для обробників черги звично обробляти повідомлення послідовно в довготривалих CLI-процесах, які не завершуються після обробки одного повідомлення. Остерігайтеся станів сервісів, щоб запобігти витоку інформації та/або памʼяті, оскільки Symfony впроваджуватиме той самий екземпляр сервісу в усіх повідомленнях, зберігаючи внутрішній стан сервісів.

Проте певні сервіси Symfony, як-от Monolog fingers crossed handler, «течуть» за задумом. Symfony надає можливість скидання сервісів (service reset), щоб розвʼязати цю проблему. Автоматично скидаючи контейнер між двома повідомленнями, Symfony шукає будь-які сервіси, що реалізують Symfony\Contracts\Service\ResetInterface (включно з вашими власними сервісами), і викликає їхній метод reset(), щоб вони могли очистити свій внутрішній стан.

Якщо сервіс не є безстановим і ви хочете скидати його властивості після кожного повідомлення, то цей сервіс має реалізовувати Symfony\Contracts\Service\ResetInterface, де ви можете скинути властивості в методі reset().

Якщо ви не хочете скидати контейнер, додайте опцію --no-reset під час запуску команди messenger:consume.

Замість того щоб повністю вимикати скидання, ви також можете налаштувати, щоб скидання відбувалося кожні N повідомлень, передавши число в опцію --no-reset. Це корисно, коли скидання після кожного повідомлення надто дороге, але повна відсутність скидання призводить до витоків памʼяті чи застарілого стану:

# скидати сервіси кожні 100 повідомлень замість після кожного
php bin/console messenger:consume async --no-reset=100

Нове у версії 8.1. Можливість передавати число в опцію --no-reset було додано в Symfony 8.1.

Власна стратегія виконання повідомлень

Нове у версії 8.1. MessageExecutionStrategyInterface було додано в Symfony 8.1.

За замовчуванням обробник черги обробляє повідомлення синхронно, використовуючи Symfony\Component\Messenger\Execution\SyncMessageExecutionStrategy.

Щоб змінити спосіб обробки повідомлень, наприклад виконувати їх паралельно, реалізуйте Symfony\Component\Messenger\Execution\MessageExecutionStrategyInterface і передайте екземпляр у конструктор Worker:

use Symfony\Component\Messenger\Envelope;
use Symfony\Component\Messenger\Execution\MessageExecutionStrategyInterface;

class MyCustomExecutionStrategy implements MessageExecutionStrategyInterface
{
    public function execute(Envelope $envelope, string $transportName, callable $onHandled): void
    {
        // власна логіка виконання
    }

    // ... реалізуйте решту методів інтерфейсу
}

Транспорт з обмеженням частоти

Інколи обробнику повідомлень потрібно обмежити частоту роботи. Налаштуйте rate limiter на транспорті (потрібен компонент RateLimiter) через опцію rate_limiter:

# config/packages/messenger.yaml
framework:
    messenger:
        transports:
            async:
                rate_limiter: your_rate_limiter_name

// config/packages/messenger.php
namespace Symfony\Component\DependencyInjection\Loader\Configurator;

return App::config([
    'framework' => [
        'messenger' => [
            'transports' => [
                'async' => [
                    'rate_limiter' => 'your_rate_limiter_name',
                ],
            ],
        ],
    ],
]);

Увага. Коли на транспорті налаштований rate limiter, після досягнення ліміту блокується весь worker. Виділіть для транспорту з обмеженням частоти окремий worker, щоб не блокувати інші транспорти.

Повторні спроби та збої

Якщо під час обробки повідомлення з транспорту виникає виняток, повідомлення автоматично відправляється назад у транспорт для нової спроби. Типово повідомлення повторюється 3 рази, після чого відкидається або відправляється у failure transport. Кожна повторна спроба також затримується, на випадок якщо збій був тимчасовим. Усе це налаштовується для кожного транспорту окремо:

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

                # конфігурація стратегії повторних спроб
                retry_strategy:
                    max_retries: 3
                    # час очікування перед першою повторною спробою (у мілісекундах)
                    delay: 1000
                    # множник, який застосовується до затримки на кожній наступній спробі
                    # (наприклад, із затримкою 1000 мс множник 2 дає: 1с, 2с, 4с, ...)
                    multiplier: 2
                    # максимально дозволена затримка незалежно від множника (0 означає без обмеження)
                    max_delay: 10000
                    # коефіцієнт випадковості (від 0 до 1.0), що додається до кожної затримки,
                    # щоб кілька невдалих повідомлень не повторювались одночасно
                    jitter: 0.1
                    # повністю замінити стратегію повторних спроб власним сервісом, який
                    # реалізує Symfony\Component\Messenger\Retry\RetryStrategyInterface
                    # service: null

// config/packages/messenger.php
namespace Symfony\Component\DependencyInjection\Loader\Configurator;

return App::config([
    'framework' => [
        'messenger' => [
            'transports' => [
                'async_priority_high' => [
                    'dsn' => env('MESSENGER_TRANSPORT_DSN'),
                ],
                'retry_strategy' => [
                    'max_retries' => 3,
                    // час очікування перед першою повторною спробою (у мілісекундах)
                    'delay' => 1000,
                    // множник, який застосовується до затримки на кожній наступній спробі
                    // (наприклад, із затримкою 1000 мс множник 2 дає: 1с, 2с, 4с, ...)
                    'multiplier' => 2,
                    // максимально дозволена затримка незалежно від множника (0 означає без обмеження)
                    'max_delay' => 0,
                    // коефіцієнт випадковості (від 0 до 1.0), що додається до кожної затримки,
                    // щоб кілька невдалих повідомлень не повторювались одночасно
                    'jitter' => 0.1,
                    // повністю замінити стратегію повторних спроб власним сервісом, який
                    // реалізує Symfony\Component\Messenger\Retry\RetryStrategyInterface
                    // 'service' => null,
                ],
            ],
        ],
    ],
]);

Порада. Якщо збої у вашому застосунку повʼязані із зовнішніми сервісами, зокрема з обмеженням частоти запитів, типового вікна повторних спроб може не вистачити. Збільште max_retries, delay, multiplier і max_delay відповідно до вашого сценарію.

Порада. Під час повторної спроби Symfony запускає подію Symfony\Component\Messenger\Event\WorkerMessageRetriedEvent, тож ви можете виконати власну логіку.

Зверніть увагу. Завдяки Symfony\Component\Messenger\Stamp\SerializedMessageStamp серіалізована форма повідомлення зберігається, що дозволяє не серіалізувати його заново, якщо повідомлення пізніше повторюється.

Як уникнути повторних спроб

Інколи обробка повідомлення провалюється так, що ви точно знаєте: це назавжди і повторювати не треба. Якщо кинути Symfony\Component\Messenger\Exception\UnrecoverableMessageHandlingException, повідомлення не повторюватиметься.

Зверніть увагу. Повідомлення, які не повторюються, все одно потраплять у налаштований failure transport. Якщо цього не потрібно, обробіть помилку самостійно і дайте обробнику завершитись успішно.

Примусові повторні спроби

Інколи обробка повідомлення має провалитись так, що ви точно знаєте: це тимчасово і спробу треба повторити. Якщо кинути Symfony\Component\Messenger\Exception\RecoverableMessageHandlingException, повідомлення повторюватиметься нескінченно, а налаштування max_retries ігноруватиметься.

Власну затримку повторної спроби в мілісекундах (наприклад, щоб узяти значення із заголовка Retry-After в HTTP-відповіді) задають аргументом retryDelay у конструкторі RecoverableMessageHandlingException.

Якщо потрібно перевизначити затримку, але все одно припинити спроби після max_retries (наприклад, коли API з обмеженням частоти відповідає заголовком Retry-After, а повторювати нескінченно ви не хочете), передайте в конструктор forceRetry: false:

throw new RecoverableMessageHandlingException(
    'Rate limited by the remote API',
    retryDelay: 5000,
    forceRetry: false,
);

Нове у версії 8.1. Аргумент forceRetry у RecoverableMessageHandlingException зʼявився у Symfony 8.1.

Збереження й повторна обробка невдалих повідомлень

Якщо повідомлення провалюється, воно повторюється кілька разів (max_retries), а потім відкидається. Щоб цього не сталося, налаштуйте failure_transport:

# config/packages/messenger.yaml
framework:
    messenger:
        # після повторних спроб повідомлення відправлятимуться у транспорт "failed"
        failure_transport: failed

        transports:
            # ... інші транспорти

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

// config/packages/messenger.php
namespace Symfony\Component\DependencyInjection\Loader\Configurator;

return App::config([
    'framework' => [
        'messenger' => [
            // після повторних спроб повідомлення відправлятимуться у транспорт "failed"
            'failure_transport' => 'failed',

            'transports' => [
                // ... інші транспорти
                'failed' => 'doctrine://default?queue_name=failed',
            ],
        ],
    ],
]);

У цьому прикладі, якщо обробка повідомлення провалиться 3 рази (типове значення max_retries), воно відправиться у транспорт failed. Хоча ви можете споживати його як звичайний транспорт через messenger:consume failed, зазвичай зручніше переглянути повідомлення у failure transport вручну й вирішити, які з них повторити:

# переглянути всі повідомлення у failure transport, типовий ліміт 50
$ php bin/console messenger:failed:show

# переглянути перші 10 повідомлень
$ php bin/console messenger:failed:show --max=10

# переглянути лише повідомлення App\Message\MyMessage
$ php bin/console messenger:failed:show --class-filter='App\Message\MyMessage'

# переглянути кількість повідомлень за класом повідомлення
$ php bin/console messenger:failed:show --stats

# переглянути деталі конкретного збою
$ php bin/console messenger:failed:show 20 -vv

# для кожного повідомлення команда запитує: повторити, пропустити чи видалити
$ php bin/console messenger:failed:retry -vv

# повторити конкретні повідомлення
$ php bin/console messenger:failed:retry 20 30 --force

# видалити повідомлення без повторної спроби
$ php bin/console messenger:failed:remove 20

# видалити повідомлення без повторних спроб і показати кожне перед видаленням
$ php bin/console messenger:failed:remove 20 30 --show-messages

# видалити всі повідомлення у failure transport
$ php bin/console messenger:failed:remove --all

# видалити лише повідомлення App\Message\MyMessage
$ php bin/console messenger:failed:remove --class-filter='App\Message\MyMessage'

Якщо повідомлення провалиться знову, воно повернеться у failure transport за звичайними правилами повторних спроб. Після досягнення максимальної кількості спроб повідомлення буде остаточно відкинуто.

Написання ідемпотентних обробників

Повідомлення може бути доставлене більш ніж один раз за звичайних умов роботи. Наприклад, worker може успішно обробити повідомлення, але впасти до того, як підтвердить його транспорту. У такому разі транспорт доставить повідомлення повторно.

Якщо дворазовий запуск обробника дає різні результати (скажімо, подвійне списання коштів у клієнта, дублікати листів або подвійне зменшення залишків на складі), механізми повторних спроб не врятують від цих помилок.

Коли це можливо, проєктуйте обробники природно ідемпотентними. Тобто кількаразовий запуск має давати той самий результат, що й одноразовий. Наприклад, встановлення абсолютного значення (як-от поточного рівня складських залишків із зовнішнього джерела) повторювати безпечно, а зменшення лічильника - ні.

Для операцій, які за своєю природою неідемпотентні, наприклад обробки платежів, додайте у повідомлення стабільний ключ ідемпотентності. Ключ має виводитись із бізнес-події, а не генеруватись у момент диспетчеризації, щоб будь-яка повторна доставка тієї самої логічної події використовувала той самий ключ:

final class ProcessPayment
{
    public function __construct(
        public readonly int $orderId,
        // idempotencyKey виводиться із замовлення, а не генерується під час диспетчеризації
        // наприклад: "payment-{orderId}-{orderVersion}"
        public readonly string $idempotencyKey,
    ) {
    }
}

В обробнику перевіряйте ключ перед виконанням операції і додайте обмеження унікальності на рівні бази даних. Перевірка допомагає уникнути зайвої роботи, а обмеження захищає від одночасних повторних доставок:

#[AsMessageHandler]
final class ProcessPaymentHandler
{
    public function __invoke(ProcessPayment $message): void
    {
        if ($this->paymentRepository->existsByIdempotencyKey($message->idempotencyKey)) {
            return;
        }

        // ... обробити платіж, зберегти запис із ключем ідемпотентності
    }
}

Зверніть увагу. UUID, згенерований у момент диспетчеризації, не підходить як ключ ідемпотентності. Якщо ту саму бізнес-подію диспетчеризують двічі (наприклад, через подвійне надсилання форми), кожна диспетчеризація згенерує інший UUID, і обидва виконання відбудуться. Ключ має лишатися стабільним для всіх диспетчеризацій однієї логічної події.

Кілька транспортів для невдалих повідомлень

Іноді одного глобального failed transport замало, бо одні повідомлення важливіші за інші. У таких випадках failure transport можна перевизначити для окремих транспортів:

# config/packages/messenger.yaml
framework:
    messenger:
        # після повторних спроб повідомлення типово відправлятимуться у транспорт "failed",
        # якщо всередині транспорту не налаштовано "failed_transport"
        failure_transport: failed_default

        transports:
            async_priority_high:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                failure_transport: failed_high_priority

            # оскільки failed transport не налаштовано, використається
            # глобально заданий "failure_transport"
            async_priority_low:
                dsn: 'doctrine://default?queue_name=async_priority_low'

            failed_default: 'doctrine://default?queue_name=failed_default'
            failed_high_priority: 'doctrine://default?queue_name=failed_high_priority'

// config/packages/messenger.php
namespace Symfony\Component\DependencyInjection\Loader\Configurator;

return App::config([
    'framework' => [
        'messenger' => [
            // після повторних спроб повідомлення типово відправлятимуться у транспорт "failed",
            // якщо всередині транспорту не налаштовано "failure_transport"
            'failure_transport' => 'failed_default',

            'transports' => [
                'async_priority_high' => [
                    'dsn' => env('MESSENGER_TRANSPORT_DSN'),
                    'failure_transport' => 'failed_high_priority',
                ],
                // оскільки failed transport не налаштовано, використається
                // глобально заданий "failure_transport"
                'async_priority_low' => [
                    'dsn' => 'doctrine://default?queue_name=async_priority_low',
                ],
                'failed_default' => 'doctrine://default?queue_name=failed_default',
                'failed_high_priority' => 'doctrine://default?queue_name=failed_high_priority',
            ],
        ],
    ],
]);

Якщо failure_transport не визначено ні глобально, ні на рівні транспорту, повідомлення відкидатимуться після вичерпання повторних спроб.

Команди для невдалих повідомлень мають необовʼязкову опцію --transport, щоб указати failure_transport, налаштований на рівні транспорту.

# переглянути всі повідомлення у транспорті "failure_transport"
$ php bin/console messenger:failed:show --transport=failure_transport

# повторити конкретні повідомлення з "failure_transport"
$ php bin/console messenger:failed:retry 20 30 --transport=failure_transport --force

# видалити повідомлення з "failure_transport" без повторної спроби
$ php bin/console messenger:failed:remove 20 --transport=failure_transport

Обробка помилок декодування

Коли повідомлення не вдається декодувати (наприклад, тому що його PHP-клас видалили або перейменували після деплою), Messenger пропускає таке повідомлення через звичайний конвеєр повторних спроб і failure transport. Щойно причину усунуто, повторна спроба з failure transport автоматично декодує повідомлення знову.

Нове у версії 8.1. До Symfony 8.1 повідомлення, які не вдалося декодувати, кидали MessageDecodingFailedException і тихо видалялися з черги. Тепер вони йдуть через retry/failure transport, тож їх можна відновити.

Налаштування транспорту

Messenger підтримує кілька типів транспортів, кожен зі своїми опціями. Опції передають у транспорт через рядок DSN або через конфігурацію.

# .env
MESSENGER_TRANSPORT_DSN=amqp://localhost/%2f/messages?auto_setup=false

# config/packages/messenger.yaml
framework:
    messenger:
        transports:
            my_transport:
                dsn: "%env(MESSENGER_TRANSPORT_DSN)%"
                options:
                    auto_setup: false

// config/packages/messenger.php
namespace Symfony\Component\DependencyInjection\Loader\Configurator;

return App::config([
    'framework' => [
        'messenger' => [
            'transports' => [
                'my_transport' => [
                    'dsn' => env('MESSENGER_TRANSPORT_DSN'),
                    'options' => ['auto_setup' => false],
                ],
            ],
        ],
    ],
]);

Опції, визначені під ключем options, мають пріоритет над тими, що задані в DSN.

Транспорт AMQP

Транспорт AMQP використовує PHP-розширення AMQP для відправлення повідомлень у черги на кшталт RabbitMQ. Установіть його командою:

$ composer require symfony/amqp-messenger

DSN транспорту AMQP може виглядати так:

# .env
MESSENGER_TRANSPORT_DSN=amqp://guest:guest@localhost:5672/%2f/messages

# або використовуйте протокол AMQPS
MESSENGER_TRANSPORT_DSN=amqps://guest:guest@localhost/%2f/messages

Якщо ви хочете використовувати AMQP із шифруванням TLS/SSL, потрібно також надати CA-сертифікат. Задайте шлях до сертифіката в налаштуванні PHP.ini amqp.cacert (наприклад, amqp.cacert = /etc/ssl/certs) або в параметрі cacert у DSN (наприклад, amqps://localhost?cacert=/etc/ssl/certs/).

Типовий порт для AMQP із шифруванням TLS/SSL - 5671, але його можна перевизначити параметром port у DSN (наприклад, amqps://localhost?cacert=/etc/ssl/certs/&port=12345).

Зверніть увагу. Типово транспорт автоматично створює всі потрібні exchange, черги та binding keys. Це можна вимкнути, але тоді частина функціональності може працювати некоректно (наприклад, відкладені черги). Щоб не створювати черги автоматично, налаштуйте транспорт із queues: [].

Зверніть увагу. Споживача транспорту AMQP можна обмежити обробкою повідомлень лише з деяких черг exchange. Див. розділ про обмеження черг.

Транспорт має низку інших опцій, зокрема способи налаштувати exchange, binding keys черг тощо. Див. документацію до Symfony\Component\Messenger\Bridge\Amqp\Transport\Connection.

Транспорт має такі опції:

• auto_setup (за замовчуванням: true): чи мають exchange і черги створюватись автоматично під час send / get.

• cacert: шлях до файлу CA-сертифіката у форматі PEM.

• cert: шлях до клієнтського сертифіката у форматі PEM.

• channel_max: задає найбільший номер каналу, який дозволяє сервер. 0 означає стандартний ліміт розширення.

• confirm_timeout: таймаут підтвердження в секундах; якщо не вказано, транспорт не чекатиме на підтвердження повідомлення. Зауважте: 0 або більше секунд. Може бути дробовим.

• connect_timeout: таймаут зʼєднання. Зауважте: 0 або більше секунд. Може бути дробовим.

• frame_max: найбільший розмір фрейму, який сервер пропонує для зʼєднання, включно із заголовком фрейму й кінцевим байтом. 0 означає стандартний ліміт розширення (залежить від типового обмеження розміру фрейму в librabbitmq).

• heartbeat: затримка heartbeat зʼєднання в секундах, якої хоче сервер. 0 означає, що сервер не хоче heartbeat. Зауважте, що librabbitmq має обмежену підтримку heartbeat: перевірка відбувається лише під час блокувальних викликів.

• host: імʼя хоста сервісу AMQP.

• key: шлях до клієнтського ключа у форматі PEM.

• login: імʼя користувача для підключення до сервісу AMQP.

• password: пароль для підключення до сервісу AMQP.

• persistent (за замовчуванням: 'false'): чи є зʼєднання постійним.

• port: порт сервісу AMQP.

• read_timeout: таймаут вхідної активності. Зауважте: 0 або більше секунд. Може бути дробовим.

• retry: (опис відсутній).

• sasl_method: (опис відсутній).

• connection_name: для власних імен зʼєднань (потрібна щонайменше версія 1.10 PHP-розширення AMQP).

• verify: увімкнути або вимкнути перевірку однорангового вузла. Якщо перевірку ввімкнено, common name у сертифікаті сервера має збігатися з імʼям сервера. Типово перевірка ввімкнена.

• vhost: віртуальний хост для роботи із сервісом AMQP.

• write_timeout: таймаут вихідної активності. Зауважте: 0 або більше секунд. Може бути дробовим.

• delay[queue_name_pattern] (за замовчуванням: delay_%exchange_name%_%routing_key%_%delay%): шаблон для створення черг.

• delay[exchange_name] (за замовчуванням: delays): імʼя exchange для відкладених і повторюваних повідомлень.

• delay[arguments] (за замовчуванням: []): додаткові аргументи для черг затримок.

• delay[daily_delay_queues] (за замовчуванням: false): якщо true, щодня створюється окрема черга затримок (до імені черги додається поточна дата) з TTL в одну добу плюс затримка повідомлення. Вмикайте це, коли для відкладених повідомлень використовуються quorum queues у RabbitMQ: на відміну від класичних черг, quorum queues не оновлюють час закінчення дії черги під час повторного оголошення, тож повідомлення з довгими затримками можуть губитися, коли черга спливає.

  Нове у версії 8.1. Опція delay[daily_delay_queues] зʼявилася у Symfony 8.1.

• queues: конфігурація черг. Можна встановити false або [], якщо не потрібно створювати ні власні черги, ні типову чергу messages.

  Нове у версії 8.1. Можливість встановити queues у false, щоб пропустити привʼязку типової черги messages, зʼявилася у Symfony 8.1.

• queues[name][arguments]: додаткові аргументи.

• queues[name][binding_arguments]: аргументи, які використовуються під час привʼязки черги.

• queues[name][binding_keys]: binding keys (якщо є), які привʼязуються до цієї черги.

• queues[name][flags] (за замовчуванням: AMQP_DURABLE): прапорці черги.

• exchange[arguments]: додаткові аргументи для exchange (наприклад, alternate-exchange).

• exchange[default_publish_routing_key]: routing key для публікації, якщо його не вказано в повідомленні.

• exchange[flags] (за замовчуванням: AMQP_DURABLE): прапорці exchange.

• exchange[name]: імʼя exchange. Порожній рядок означає типовий exchange.

• exchange[type] (за замовчуванням: fanout): тип exchange.

• exchange[bindings]: вихідні exchange, до яких привʼязується цей exchange, з ключем за іменем. Напрямок привʼязки: вихідний exchange -> цей exchange.

• exchange[bindings][binding_keys]: binding/routing keys, які використовуються для привʼязки.

• exchange[bindings][binding_arguments]: додаткові аргументи привʼязки.

Специфічні для AMQP налаштування можна задати й на самому повідомленні, додавши в Envelope Symfony\Component\Messenger\Bridge\Amqp\Transport\AmqpStamp:

use Symfony\Component\Messenger\Bridge\Amqp\Transport\AmqpStamp;
// ...

$attributes = [];
$bus->dispatch(new SmsNotification(), [
    new AmqpStamp('custom-routing-key', AMQP_NOPARAM, $attributes),
]);

Транспорт AMQP автоматично додає до повідомлень Symfony\Component\Messenger\Stamp\TransportMessageIdStamp під час відправлення й отримання. Цей stamp зберігає ID повідомлення AMQP, що покращує контекст логів, коли повідомлення провалюються й повторюються.

Увага. Споживачі не відображаються в адмін-панелі, бо цей транспорт не покладається на блокувальний \AmqpQueue::consume(). Блокувальний отримувач робить неефективними опції --time-limit/--memory-limit команди messenger:consume, а також команду messenger:stop-workers, бо всі вони спираються на те, що отримувач повертає керування негайно, незалежно від того, знайшов він повідомлення чи ні. Worker сам ітерує, доки не отримає повідомлення для обробки та/або доки не спрацює одна з умов зупинки. Тому логіка зупинки worker недосяжна, якщо він застряг у блокувальному виклику.

Порада. Якщо ваш застосунок стикається з винятками сокетів або high connection churn (це видно з швидкого створення й видалення зʼєднань), розгляньте AMQProxy. Цей інструмент працює як шлюз між Symfony Messenger і сервером AMQP, підтримує стабільні зʼєднання й мінімізує накладні витрати (що також покращує загальну продуктивність).

Транспорт Doctrine

Транспорт Doctrine дозволяє зберігати повідомлення в таблиці бази даних. Установіть його командою:

$ composer require symfony/doctrine-messenger

DSN транспорту Doctrine може виглядати так:

# .env
MESSENGER_TRANSPORT_DSN=doctrine://default

Формат - doctrine://<connection_name>, на випадок якщо у вас кілька зʼєднань і ви хочете використати не "default". Транспорт автоматично створить таблицю з іменем messenger_messages.

Щоб змінити типове імʼя таблиці, передайте власне імʼя в DSN через опцію table_name:

# .env
MESSENGER_TRANSPORT_DSN=doctrine://default?table_name=your_custom_table_name

Типово таблиця Doctrine створюється автоматично під час диспетчеризації першого повідомлення. Це зручно в розробці, але у продакшені краще встановити auto_setup у false в MESSENGER_TRANSPORT_DSN. Тоді згенеруйте міграцію і створюйте таблицю явно як частину процесу деплою.

Транспорт має такі опції:

• table_name (за замовчуванням: messenger_messages): імʼя таблиці.

• queue_name (за замовчуванням: default): імʼя черги (колонка в таблиці, щоб використовувати одну таблицю для кількох транспортів).

  Увага. Якщо ви використовуєте PostgreSQL з увімкненою опцією use_notify, наполегливо рекомендуємо, щоб кілька транспортів не мали однакового значення queue_name (включно зі значенням default). Спільне імʼя черги для різних транспортів суттєво знижує ефективність LISTEN/NOTIFY у виявленні майбутніх відкладених повідомлень.

• redeliver_timeout (за замовчуванням: 3600): таймаут у секундах перед повторною спробою повідомлення, яке є в черзі, але в стані "handling" (таке трапляється, якщо worker з якоїсь причини зупинився; врешті повідомлення треба повторити).

  Зверніть увагу. Встановлюйте redeliver_timeout більшим за найдовшу тривалість обробки повідомлення. Інакше частина повідомлень почне оброблятися вдруге, поки перша обробка ще триває.

• auto_setup: чи має таблиця створюватись автоматично під час send / get.

Для PostgreSQL доступні додаткові опції, які задіюють можливість LISTEN/NOTIFY. Це продуктивніший підхід, ніж типове опитування (polling) транспорту Doctrine, бо PostgreSQL сам повідомляє workerʼам, що в таблицю вставлено нове повідомлення.

• use_notify (за замовчуванням: true): чи використовувати LISTEN/NOTIFY.
• check_delayed_interval (за замовчуванням: 60000): інтервал перевірки відкладених повідомлень у мілісекундах. Встановіть 0, щоб вимкнути перевірки.
• get_notify_timeout (за замовчуванням: 60000): максимальний час у мілісекундах, який PostgreSqlNotifyOnIdleListener чекає на NOTIFY, поки worker простоює. Встановіть 0, щоб вимкнути очікування.

PostgreSqlConnection::get() ніколи не чекає на сповіщення. Коли жоден idle listener не активний, метод виконує команду LISTEN, збирає сповіщення, які вже надійшли, і повертає керування. Завдяки цьому worker, який споживає з кількох черг на PostgreSQL (або з суміші PostgreSQL та інших транспортів), на кожній ітерації циклу продовжує опитувати всі транспорти в порядку пріоритету, а не зависає на одному.

За очікування під час простою worker відповідає підписник подій Symfony\Component\Messenger\Bridge\Doctrine\EventListener\PostgreSqlNotifyOnIdleListener, який реєструється автоматично при використанні транспорту Doctrine з PostgreSQL. Слухач вбудовується в життєвий цикл worker:

• На WorkerStartedEvent: перевіряє, що всі транспорти PostgreSQL використовують те саме зʼєднання DBAL і ту саму таблицю, після чого реєструє LISTEN лише на першому зʼєднанні. Решта позначають get() як оброблений ззовні, щоб не накопичувати непрочитані сповіщення.

• На WorkerRunningEvent (простій): блокується на waitForNotify() з активним зʼєднанням і розумним таймаутом, обмеженим:

  • опцією транспорту get_notify_timeout;
  • найближчим відкладеним повідомленням серед усіх черг PostgreSQL, щоб worker прокинувся вчасно для його обробки;
  • дедлайном worker (--time-limit);
  • тривалістю сну worker, якщо споживаються також не-PostgreSQL транспорти, щоб їх регулярно опитували.

Таким чином головний цикл worker перевіряє всі транспорти в порядку пріоритету на кожній ітерації і блокується лише після того, як усі черги виявилися порожніми.

Увага. Коли worker споживає з кількох транспортів PostgreSQL, усі вони мають використовувати те саме зʼєднання DBAL і те саме table_name. Інакше під час запуску worker буде кинуто LogicException.

Нове у версії 8.1. PostgreSqlNotifyOnIdleListener зʼявився у Symfony 8.1.

Транспорт Doctrine підтримує опцію --keepalive, періодично оновлюючи мітку часу delivered_at, щоб повідомлення не доставлялося повторно.

Транспорт Beanstalkd

Транспорт Beanstalkd відправляє повідомлення напряму в робочу чергу Beanstalkd. Установіть його командою:

$ composer require symfony/beanstalkd-messenger

DSN транспорту Beanstalkd може виглядати так:

# .env
MESSENGER_TRANSPORT_DSN=beanstalkd://localhost:11300?tube_name=foo&timeout=4&ttr=120

# Якщо порт не вказано, типово використовується 11300
MESSENGER_TRANSPORT_DSN=beanstalkd://localhost

Транспорт має такі опції:

• bury_on_reject (за замовчуванням: false): якщо true, відхилені повідомлення переводяться в Beanstalkd у стан "buried" замість видалення.
• timeout (за замовчуванням: 0): таймаут резервування повідомлення в секундах. 0 змусить сервер повернути керування негайно - або з відповіддю, або кинувши TransportException.
• ttr (за замовчуванням: 90): час на виконання повідомлення в секундах, після якого воно повертається в чергу готових.
• tube_name (за замовчуванням: default): імʼя черги.

Транспорт Beanstalkd підтримує опцію --keepalive, використовуючи команду Beanstalkd touch, щоб періодично скидати ttr завдання.

Транспорт Beanstalkd дозволяє задати пріоритет повідомлень, які диспетчеризуються. Використайте Symfony\Component\Messenger\Bridge\Beanstalkd\Transport\BeanstalkdPriorityStamp і передайте число з пріоритетом (за замовчуванням 1024; менші числа означають вищий пріоритет):

use App\Message\SomeMessage;
use Symfony\Component\Messenger\Stamp\BeanstalkdPriorityStamp;

$this->bus->dispatch(new SomeMessage('some data'), [
    // 0 = найвищий пріоритет
    // 2**32 - 1 = найнижчий пріоритет
    new BeanstalkdPriorityStamp(0),
]);

Транспорт Redis

Транспорт Redis використовує streams для черги повідомлень. Йому потрібні PHP-розширення Redis (>=4.3) і запущений сервер Redis (^5.0). Установіть його командою:

$ composer require symfony/redis-messenger

DSN транспорту Redis може виглядати так:

# .env
MESSENGER_TRANSPORT_DSN=redis://localhost:6379/messages
# Приклад повного DSN
MESSENGER_TRANSPORT_DSN=redis://password@localhost:6379/messages/symfony/consumer?auto_setup=true&serializer=1&stream_max_entries=0&dbindex=0
# Приклад Redis Cluster
MESSENGER_TRANSPORT_DSN=redis://host-01:6379,redis://host-02:6379,redis://host-03:6379,redis://host-04:6379
# Redis Cluster з автоматичним виявленням вузлів
MESSENGER_TRANSPORT_DSN=redis://localhost:6379/messages?cluster=true
# Приклад Unix Socket
MESSENGER_TRANSPORT_DSN=redis:///var/run/redis.sock
# Приклад TLS
MESSENGER_TRANSPORT_DSN=rediss://localhost:6379/messages
# Приклад із кількома хостами Redis Sentinel
MESSENGER_TRANSPORT_DSN=redis:?host[redis1:26379]&host[redis2:26379]&host[redis3:26379]&sentinel_master=db
# Redis Sentinel з окремими обліковими даними для master і sentinel
MESSENGER_TRANSPORT_DSN=redis://master-pass@?host[redis1:26379]&host[redis2:26379]&sentinel_master=db&auth=sentinel-pass
# Redis Sentinel з обліковими даними ACL і для master, і для sentinel
MESSENGER_TRANSPORT_DSN=redis://master-user:master-pass@?host[redis1:26379]&host[redis2:26379]&sentinel_master=db&auth[]=sentinel-user&auth[]=sentinel-pass

Низку опцій можна налаштувати через DSN або через ключ options під транспортом у messenger.yaml:

• stream (за замовчуванням: messages): імʼя Redis stream.

• group (за замовчуванням: symfony): імʼя групи споживачів Redis.

• consumer (за замовчуванням: consumer): імʼя споживача в Redis. Дозволяє задати явний ідентифікатор споживача. Рекомендовано в середовищах із кількома workerʼами, щоб уникнути дублювання обробки повідомлень. Зазвичай задається через змінну середовища:

  # config/packages/messenger.yaml
  framework:
      messenger:
          transports:
              redis:
                  dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                  options:
                      consumer: '%env(MESSENGER_CONSUMER_NAME)%'
  

• auto_setup (за замовчуванням: true): чи створювати групу Redis автоматично.

• auth: пароль Redis. Використовуйте рядок для автентифікації лише за паролем або масив із двох елементів [username, password] для автентифікації на основі ACL. При використанні Redis Sentinel ця опція задає облікові дані sentinel. Облікові дані master передаються через userinfo в DSN (наприклад, redis://master-pass@host).

• delete_after_ack (за замовчуванням: true): якщо true, повідомлення видаляються автоматично після обробки.

• delete_after_reject (за замовчуванням: true): якщо true, повідомлення видаляються автоматично, якщо їх відхилено.

• lazy (за замовчуванням: false): підключатися лише тоді, коли зʼєднання справді потрібне.

• serializer (за замовчуванням: Redis::SERIALIZER_PHP): як серіалізувати кінцевий payload у Redis (опція Redis::OPT_SERIALIZER).

• stream_max_entries (за замовчуванням: 0): максимальна кількість записів, до якої обрізатиметься stream. Встановіть достатньо велике число, щоб не втратити повідомлення, які ще очікують на обробку.

• redeliver_timeout (за замовчуванням: 3600): таймаут у секундах перед повторною спробою повідомлення, що очікує й належить покинутому споживачу (таке трапляється, якщо worker з якоїсь причини помер; врешті повідомлення треба повторити).

• claim_interval (за замовчуванням: 60000): інтервал у мілісекундах, з яким перевіряються повідомлення в очікуванні чи від покинутих споживачів, щоб їх перебрати.

• persistent_id (за замовчуванням: null): рядок; якщо null, зʼєднання непостійне.

• retry_interval (за замовчуванням: 0): ціле число, значення в мілісекундах.

• read_timeout (за замовчуванням: 0): дробове число, значення в секундах; типове значення означає без обмеження.

• timeout (за замовчуванням: 0): таймаут зʼєднання. Дробове число, значення в секундах; типове значення означає без обмеження.

• sentinel_master (за замовчуванням: null): рядок; якщо null або порожній, підтримка Sentinel вимкнена.

• redis_sentinel (за замовчуванням: null): псевдонім опції sentinel_master.

• cluster (за замовчуванням: false): якщо true, примусово використовується RedisCluster замість Redis. Це корисно при підключенні до Redis Cluster, коли ви хочете автоматичне виявлення вузлів, а не перелік усіх вузлів кластера в DSN.

  Нове у версії 8.1. Опція cluster зʼявилася у Symfony 8.1.

• ssl (за замовчуванням: null): набір SSL context options для TLS-каналу. Корисно, наприклад, щоб змінити вимоги до TLS-каналу в тестах:

  # config/packages/messenger.yaml
  framework:
      when@test:
          messenger:
              transports:
                  redis:
                      dsn: "rediss://localhost"
                      options:
                          ssl:
                              allow_self_signed: true
                              capture_peer_cert: true
                              capture_peer_cert_chain: true
                              disable_compression: true
                              SNI_enabled: true
                              verify_peer: true
                              verify_peer_name: true
  

  // config/packages/messenger.php
  namespace Symfony\Component\DependencyInjection\Loader\Configurator;
  
  return App::config([
      'when@test' => [
          'framework' => [
              'messenger' => [
                  'transports' => [
                      'redis' => [
                          'dsn' => "rediss://localhost",
                          'options' => [
                              'ssl' => [
                                  'allow_self_signed' => true,
                                  'capture_peer_cert' => true,
                                  'capture_peer_cert_chain' => true,
                                  'disable_compression' => true,
                                  'SNI_enabled' => true,
                                  'verify_peer' => true,
                                  'verify_peer_name' => true,
                              ],
                          ],
                      ],
                  ],
              ],
          ],
      ],
  ]);
  

Увага. Ніколи не має бути більше однієї запущеної команди messenger:consume з однаковою комбінацією stream, group і consumer, інакше повідомлення можуть оброблятися кілька разів. Якщо ви запускаєте кілька обробників черги, задайте consumer через змінну середовища, наприклад %env(MESSENGER_CONSUMER_NAME)%, яку встановлює Supervisor (приклад нижче) або будь-який інший сервіс керування процесами worker.

Порада. У контейнерному середовищі як імʼя споживача можна використати HOSTNAME, бо на контейнер/хост припадає лише один worker. Якщо контейнерами керує Kubernetes, розгляньте StatefulSet, щоб мати стабільні імена.

Порада. Встановіть delete_after_ack у true (якщо ви використовуєте одну групу) або задайте stream_max_entries (якщо можете оцінити прийнятну максимальну кількість записів у вашому випадку), щоб уникнути витоків памʼяті. Інакше всі повідомлення залишатимуться в Redis назавжди.

Транспорт Redis підтримує опцію --keepalive, використовуючи команду Redis XCLAIM, щоб періодично скидати час простою повідомлення до нуля.

Перегляд і пошук повідомлень

Транспорт Redis реалізує Symfony\Component\Messenger\Transport\Receiver\ListableReceiverInterface, що дозволяє переглядати й шукати повідомлення в Redis stream. Це зручно для інспекції черги або побудови власних інструментів адміністрування.

Доступ до цих можливостей надає отримувач:

use Symfony\Component\Messenger\Transport\Receiver\ListableReceiverInterface;

// отримати receiver для вашого транспорту Redis
$receiver = $container->get('messenger.receiver.your_transport_name');

if ($receiver instanceof ListableReceiverInterface) {
    // перелічити всі повідомлення у stream
    foreach ($receiver->all() as $envelope) {
        // $envelope є екземпляром Symfony\Component\Messenger\Envelope
    }

    // перелічити повідомлення з обмеженням
    foreach ($receiver->all(10) as $envelope) {
        // обробити до 10 повідомлень
    }

    // знайти конкретне повідомлення за його ID у Redis stream
    $envelope = $receiver->find('1234567890-0');
    if (null !== $envelope) {
        // $envelope є екземпляром Symfony\Component\Messenger\Envelope
    }
}

Зверніть увагу. Метод all() повертає всі повідомлення, які наразі є в Redis stream, включно з тими, що зараз обробляються. Кожен повернутий envelope містить Symfony\Component\Messenger\Bridge\Redis\Transport\RedisReceivedStamp з ID повідомлення Redis і Symfony\Component\Messenger\Stamp\TransportMessageIdStamp для внутрішнього відстеження.

Нове у версії 8.1. Підтримка ListableReceiverInterface у RedisReceiver зʼявилася у Symfony 8.1.

Транспорт In Memory

Транспорт in-memory насправді не доставляє повідомлень. Замість цього він тримає їх у памʼяті протягом запиту, що зручно для тестування. Наприклад, якщо у вас є транспорт async_priority_normal, ви можете перевизначити його в середовищі test на цей транспорт:

# config/packages/messenger.yaml
when@test:
    framework:
        messenger:
            transports:
                async_priority_normal: 'in-memory://'

// config/packages/messenger.php
namespace Symfony\Component\DependencyInjection\Loader\Configurator;

return App::config([
    'when@test' => [
        'framework' => [
            'messenger' => [
                'transports' => [
                    'async_priority_normal' => 'in-memory://',
                ],
            ],
        ],
    ],
]);

Тоді під час тестування повідомлення не доставлятимуться в реальний транспорт. Ба більше, у тесті можна перевірити, що за запит було відправлено рівно одне повідомлення:

// tests/Controller/DefaultControllerTest.php
namespace App\Tests\Controller;

use Symfony\Bundle\FrameworkBundle\Test\WebTestCase;
use Symfony\Component\Messenger\Transport\InMemory\InMemoryTransport;

class DefaultControllerTest extends WebTestCase
{
    public function testSomething(): void
    {
        $client = static::createClient();
        // ...

        $this->assertSame(200, $client->getResponse()->getStatusCode());

        /** @var InMemoryTransport $transport */
        $transport = $this->getContainer()->get('messenger.transport.async_priority_normal');
        $this->assertCount(1, $transport->getSent());
    }
}

Транспорт має такі опції:

• serialize (boolean, за замовчуванням: false) - чи серіалізувати повідомлення. Це корисно, щоб перевірити додатковий шар, особливо коли ви використовуєте власний серіалізатор повідомлень.

Примітка. Усі транспорти in-memory скидаються автоматично після кожного тесту в тестових класах, що успадковують Symfony\Bundle\FrameworkBundle\Test\KernelTestCase або Symfony\Bundle\FrameworkBundle\Test\WebTestCase.

Для обробки повідомлень можна застосувати дві взаємодоповнювальні стратегії тестування. Перша: тестувати обробники як звичайні PHP-класи, підставляючи моки й викликаючи __invoke() напряму, без жодної інфраструктури Messenger. Так перевіряють бізнес-логіку, класифікацію винятків і граничні випадки:

// tests/MessageHandler/SendInvoiceHandlerTest.php
final class SendInvoiceHandlerTest extends TestCase
{
    public function testThrowsUnrecoverableForUnknownOrder(): void
    {
        $repository = $this->createMock(OrderRepository::class);
        $repository->method('find')->willReturn(null);

        $handler = new SendInvoiceHandler($repository);

        $this->expectException(UnrecoverableMessageHandlingException::class);
        $handler(new SendInvoice(orderId: 99));
    }
}

Друга: використати транспорт in-memory:// у функціональних тестах, щоб перевірити, що потрібне повідомлення відправлено в очікуваний транспорт. Замапте кожен іменований транспорт на in-memory:// у тестовій конфігурації та перевіряйте конкретний транспорт:

// tests/Controller/CheckoutControllerTest.php
/** @var InMemoryTransport $transport */
$transport = self::getContainer()->get('messenger.transport.orders_high');

self::assertCount(1, $transport->getSent());
self::assertInstanceOf(
    SendOrderConfirmation::class,
    $transport->getSent()[0]->getMessage()
);

Amazon SQS

Транспорт Amazon SQS ідеально підходить для застосунків, розміщених в AWS. Встановіть його командою:

composer require symfony/amazon-sqs-messenger

DSN транспорту SQS може виглядати так:

# .env
MESSENGER_TRANSPORT_DSN=https://sqs.eu-west-3.amazonaws.com/123456789012/messages?access_key=AKIAIOSFODNN7EXAMPLE&secret_key=j17M97ffSVoKI0briFoo9a
MESSENGER_TRANSPORT_DSN=sqs://localhost:9494/messages?sslmode=disable

Примітка. Транспорт автоматично створить потрібні черги. Це можна вимкнути, задавши опції auto_setup значення false.

Порада. Перед надсиланням або отриманням повідомлення Symfony має перетворити назву черги на URL черги AWS, викликавши API GetQueueUrl в AWS. Цього додаткового виклику API можна уникнути, якщо передати DSN, який і є URL черги.

Транспорт має низку опцій:

• access_key - ключ доступу AWS (має бути urlencoded)
• account (за замовчуванням: власник облікових даних) - ідентифікатор облікового запису AWS
• auto_setup (за замовчуванням: true) - чи потрібно створювати чергу автоматично під час надсилання / отримання.
• buffer_size (за замовчуванням: 9) - кількість повідомлень для попереднього вибирання
• debug (за замовчуванням: false) - якщо true, логує всі HTTP-запити й відповіді (це впливає на продуктивність)
• delete_on_rejection (за замовчуванням: true) - якщо задано false, відхилене повідомлення не буде видалене. Натомість зміниться його видимість, щоб SQS міг опрацювати повторні спроби
• endpoint (за замовчуванням: https://sqs.eu-west-1.amazonaws.com) - абсолютний URL сервісу SQS
• poll_timeout (за замовчуванням: 0.1) - тривалість очікування нового повідомлення в секундах
• queue_name (за замовчуванням: messages) - назва черги
• queue_attributes - атрибути черги згідно з SQS CreateQueue API. Масив рядків, індексований ключами AsyncAws\Sqs\Enum\QueueAttributeName.
• queue_tags - теги розподілу витрат для черги згідно з SQS CreateQueue API. Масив рядків, індексований рядками.
• region (за замовчуванням: eu-west-1) - назва регіону AWS
• retry_delay (за замовчуванням: 0) - використовується лише тоді, коли delete_on_rejection має значення false. Визначає таймаут видимості (у секундах), який SQS має застосувати, коли повідомлення відхилено
• secret_key - секретний ключ AWS
• session_token - токен сесії AWS
• visibility_timeout (за замовчуванням: конфігурація черги) - кількість секунд, протягом яких повідомлення буде невидимим (Visibility Timeout)
• wait_time (за замовчуванням: 20) - тривалість long polling у секундах

Примітка. Параметр wait_time визначає максимальний час, який Amazon SQS має чекати на появу повідомлення в черзі, перш ніж надіслати відповідь. Він допомагає зменшити вартість використання Amazon SQS, скорочуючи кількість порожніх відповідей.

Параметр poll_timeout визначає час, який отримувач має чекати перед поверненням null. Він не дає заблокувати виклик інших отримувачів.

Примітка. Якщо назва черги має суфікс .fifo, AWS створить FIFO-чергу (FIFO queue). Використовуйте марку Symfony\Component\Messenger\Bridge\AmazonSqs\Transport\AmazonSqsFifoStamp, щоб задати Message group ID і Message deduplication ID.

Інша можливість - увімкнути Symfony\Component\Messenger\Bridge\AmazonSqs\Middleware\AddFifoStampMiddleware. Якщо ваше повідомлення реалізує Symfony\Component\Messenger\Bridge\AmazonSqs\MessageDeduplicationAwareInterface, middleware автоматично додасть Symfony\Component\Messenger\Bridge\AmazonSqs\Transport\AmazonSqsFifoStamp і задасть Message deduplication ID.

Крім того, якщо ваше повідомлення реалізує Symfony\Component\Messenger\Bridge\AmazonSqs\MessageGroupAwareInterface, middleware автоматично задасть Message group ID марки.

Коли ви не задаєте ці ідентифікатори, транспорт надсилає власні значення за замовчуванням: однаковий Message group ID для кожного повідомлення, тож черга обробляє їх по одному, і Message deduplication ID, побудований з хешу заголовків і тіла повідомлення, тож черга вважає два однакові повідомлення дублікатами.

Докладніше про middleware можна прочитати у відповідному розділі (the dedicated section).

FIFO-черги не підтримують затримки для окремого повідомлення, у налаштуваннях стратегії повторних спроб потрібне значення delay: 0.

Увага! Дедуплікація повідомлень в AWS SQS базується на часі, а не на черзі. Щойно Message deduplication ID було використано, SQS відхиляє будь-яке повідомлення, надіслане з тим самим ID, протягом наступних 5 хвилин, незалежно від того, чи оригінальне повідомлення вже спожите або видалене з черги. Подробиці - у документації SQS message deduplication.

Транспорт SQS підтримує опцію --keepalive, використовуючи дію ChangeMessageVisibility для періодичного оновлення VisibilityTimeout повідомлення.

Серіалізація повідомлень

Коли повідомлення надсилаються в транспорт (і отримуються з нього), вони серіалізуються нативними функціями PHP serialize() і unserialize(). Це можна змінити глобально (або для кожного транспорту окремо) на сервіс, що реалізує Symfony\Component\Messenger\Transport\Serialization\SerializerInterface:

# config/packages/messenger.yaml
framework:
    messenger:
        serializer:
            default_serializer: messenger.transport.symfony_serializer
            symfony_serializer:
                format: json
                context: { }

        transports:
            async_priority_normal:
                dsn: # ...
                serializer: messenger.transport.symfony_serializer

// config/packages/messenger.php
namespace Symfony\Component\DependencyInjection\Loader\Configurator;

return App::config([
    'framework' => [
        'messenger' => [
            'serializer' => [
                'default_serializer' => 'messenger.transport.symfony_serializer',
                'symfony_serializer' => [
                    'format' => 'json',
                    'context' => [],
                ],
            ],
            'transports' => [
                'async_priority_normal' => [
                    'dsn' => // ...,
                    'serializer' => 'messenger.transport.symfony_serializer',
                ],
            ],
        ],
    ],
]);

messenger.transport.symfony_serializer - це вбудований сервіс, який використовує компонент Serializer і який можна налаштувати кількома способами. Якщо ви таки вирішите користуватися серіалізатором Symfony, ви можете керувати контекстом у кожному окремому випадку через Symfony\Component\Messenger\Stamp\SerializerStamp (див. розділ «Конверти й марки»).

Порада. Коли ви надсилаєте повідомлення іншому застосунку або отримуєте їх від нього, вам може знадобитися більший контроль над процесом серіалізації. Власний серіалізатор дає такий контроль. Подробиці - у туторіалі SymfonyCasts про серіалізатор повідомлень (SymfonyCasts' message serializer tutorial).

Налаштування заголовка type

Спілкуючись із застосунками не на Symfony або коли структура класів повідомлень у застосунках відрізняється, вам може знадобитися змінити заголовок type серіалізованих повідомлень. За замовчуванням цей заголовок містить повне імʼя класу (FQCN) класу повідомлення.

Використайте опцію serializedTypeName атрибута Symfony\Component\Messenger\Attribute\AsMessage, щоб визначити власну назву типу:

// src/Message/OrderPlaced.php
namespace App\Message;

use Symfony\Component\Messenger\Attribute\AsMessage;

#[AsMessage(serializedTypeName: 'order.placed')]
class OrderPlaced
{
    public function __construct(
        public string $orderId,
    ) {
    }
}

Коли це повідомлення буде серіалізовано, заголовок type міститиме order.placed замість App\Message\OrderPlaced.

Нове у версії 8.1. Опцію serializedTypeName було додано в Symfony 8.1.

Закриття зʼєднань

Коли ви використовуєте транспорт, який потребує зʼєднання, ви можете закрити його викликом методу Symfony\Component\Messenger\Transport\CloseableTransportInterface::close, щоб звільнити ресурси в довготривалих процесах.

Цей інтерфейс реалізують такі транспорти: AmazonSqs, Amqp і Redis. Якщо вам потрібно закрити зʼєднання Doctrine, це можна зробити за допомогою middleware (using middleware).

Запуск команд і зовнішніх процесів

Запуск команди

Будь-яку команду можна запустити, відправивши Symfony\Component\Console\Messenger\RunCommandMessage. Symfony подбає про обробку цього повідомлення й виконає команду, передану в параметрі повідомлення:

use Symfony\Component\Console\Messenger\RunCommandMessage;
use Symfony\Component\Messenger\MessageBusInterface;

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

    public function cleanUp(): void
    {
        // Довге завдання з якимось кешуванням...

        // Коли воно завершилось, відправляємо кілька команд очищення
        $this->bus->dispatch(new RunCommandMessage('app:my-cache:clean-up --dir=var/temp'));
        $this->bus->dispatch(new RunCommandMessage('cache:clear'));
    }
}

Ви можете налаштувати поведінку на випадок, коли під час виконання команди щось піде не так. Для цього скористайтеся параметрами throwOnFailure і catchExceptions під час створення екземпляра Symfony\Component\Console\Messenger\RunCommandMessage.

Після обробки обробник поверне Symfony\Component\Console\Messenger\RunCommandContext, який містить чимало корисної інформації, як-от код виходу або вивід процесу. Докладніше - на сторінці, присвяченій результатам обробників (handler results).

Запуск зовнішнього процесу

Messenger має зручний помічник для запуску зовнішніх процесів через відправлення повідомлення. Він використовує можливості компонента Process. Коли ви відправите Symfony\Component\Process\Messenger\RunProcessMessage, Messenger подбає про створення нового процесу з переданими вами параметрами:

use Symfony\Component\Messenger\MessageBusInterface;
use Symfony\Component\Process\Messenger\RunProcessMessage;

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

    public function cleanUp(): void
    {
        $this->bus->dispatch(new RunProcessMessage(['rm', '-rf', 'var/log/temp/*'], cwd: '/my/custom/working-dir'));

        // ...
    }
}

Якщо ви хочете скористатися можливостями shell, як-от перенаправлення чи пайпи, використайте статичний фабричний метод Symfony\Component\Process\Messenger\RunProcessMessage::fromShellCommandline:

use Symfony\Component\Messenger\MessageBusInterface;
use Symfony\Component\Process\Messenger\RunProcessMessage;

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

    public function cleanUp(): void
    {
        $this->bus->dispatch(RunProcessMessage::fromShellCommandline('echo "Hello World" > var/log/hello.txt'));

        // ...
    }
}

Докладніше про це - у документації про використання можливостей shell операційної системи (using features from the OS shell).

Після обробки обробник поверне Symfony\Component\Process\Messenger\RunProcessContext, який містить чимало корисної інформації, як-от код виходу або вивід процесу. Докладніше - на сторінці, присвяченій результатам обробників (handler results).

Захист повідомлень підписами

Коли повідомлення надсилаються в черги повідомлень, існує потенційний ризик безпеки: зловмисник може підкинути в чергу підроблені дані. Хоча черги повідомлень мають бути належно захищені від несанкціонованого доступу, Symfony додає ще один рівень захисту, підтримуючи підписування повідомлень.

Це особливо важливо для обробників, які виконують команди чи процеси, тому в RunProcessHandler підписування повідомлень увімкнене за замовчуванням.

Увімкнення підписування повідомлень

Щоб увімкнути підписування повідомлень для вашого обробника, задайте опції sign значення true:

// src/MessageHandler/SmsNotificationHandler.php
namespace App\MessageHandler;

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

#[AsMessageHandler(sign: true)]
class SmsNotificationHandler
{
    public function __invoke(SmsNotification $message): void
    {
        // ... обробка повідомлення
    }
}

# config/services.yaml
services:
    App\MessageHandler\SmsNotificationHandler:
        tags:
            - { name: messenger.message_handler, sign: true }

// config/services.php
use App\MessageHandler\SmsNotificationHandler;

$container->register(SmsNotificationHandler::class)
    ->addTag('messenger.message_handler', ['sign' => true]);

Коли підписування увімкнене:

1. Повідомлення підписуються HMAC-підписом, обчисленим із секретним ключем вашого застосунку (параметр kernel.secret).
2. Підпис додається до заголовків повідомлення (Body-Sign і Sign-Algo), коли повідомлення надсилається в транспорт.
3. Коли повідомлення отримано й декодовано, підпис автоматично перевіряється.
4. Якщо підпису немає або він недійсний, буде викинуто Symfony\Component\Messenger\Exception\InvalidMessageSignatureException, і повідомлення не буде оброблене.

Пінг вебсервісу

Іноді потрібно регулярно пінгувати вебсервіс, щоб дізнатися його стан, наприклад чи він працює. Це можна зробити, відправивши Symfony\Component\HttpClient\Messenger\PingWebhookMessage:

use Symfony\Component\HttpClient\Messenger\PingWebhookMessage;
use Symfony\Component\Messenger\MessageBusInterface;

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

    public function ping(): void
    {
        // На 3xx/4xx/5xx викидається HttpExceptionInterface
        $this->bus->dispatch(new PingWebhookMessage('GET', 'https://example.com/status'));

        // Пінг, але без викидання винятку на 3xx/4xx/5xx
        $this->bus->dispatch(new PingWebhookMessage('GET', 'https://example.com/status', throw: false));

        // Можна використати будь-яку дійсну опцію HttpClientInterface
        $this->bus->dispatch(new PingWebhookMessage('POST', 'https://example.com/status', [
            'headers' => [
                'Authorization' => 'Bearer ...'
            ],
            'json' => [
                'data' => 'some-data',
            ],
        ]));
    }
}

Обробник поверне Symfony\Contracts\HttpClient\ResponseInterface, що дає змогу зібрати й опрацювати інформацію, яку повернув HTTP-запит.

Отримання результатів від ваших обробників

Коли повідомлення оброблене, Symfony\Component\Messenger\Middleware\HandleMessageMiddleware додає Symfony\Component\Messenger\Stamp\HandledStamp для кожного обʼєкта, що обробив повідомлення. Так ви можете отримати значення, яке повернув обробник (чи обробники):

use Symfony\Component\Messenger\MessageBusInterface;
use Symfony\Component\Messenger\Stamp\HandledStamp;

$envelope = $messageBus->dispatch(new SomeMessage());

// отримати значення, яке повернув останній обробник повідомлення
$handledStamp = $envelope->last(HandledStamp::class);
$handledStamp->getResult();

// або отримати інформацію про всі обробники
$handledStamps = $envelope->all(HandledStamp::class);

Отримання результатів під час роботи з шинами команд і запитів

Компонент Messenger можна використовувати в CQRS-архітектурах, де шини команд і запитів є центральними елементами застосунку. Прочитайте статтю Мартіна Фаулера про CQRS (article about CQRS), щоб дізнатися більше, а також про те, як налаштувати кілька шин (how to configure multiple buses).

Оскільки запити зазвичай синхронні й мають бути оброблені один раз, отримання результату від обробника - поширена потреба.

Для отримання результату обробника під час синхронної обробки існує Symfony\Component\Messenger\HandleTrait. Він також гарантує, що зареєстровано рівно один обробник. HandleTrait можна використовувати в будь-якому класі, що має властивість $messageBus:

// src/Action/ListItems.php
namespace App\Action;

use App\Message\ListItemsQuery;
use App\MessageHandler\ListItemsQueryResult;
use Symfony\Component\DependencyInjection\Attribute\Target;
use Symfony\Component\Messenger\HandleTrait;
use Symfony\Component\Messenger\MessageBusInterface;

class ListItems
{
    use HandleTrait;

    // HandleTrait потребує властивості з іменем $messageBus
    public function __construct(
        #[Target('query.bus')] private MessageBusInterface $messageBus,
    ) {
    }

    public function __invoke(): void
    {
        $result = $this->query(new ListItemsQuery(/* ... */));

        // Зробити щось із результатом
        // ...
    }

    // Створювати такий метод необовʼязково, але він дозволяє вказати тип результату
    private function query(ListItemsQuery $query): ListItemsQueryResult
    {
        return $this->handle($query);
    }
}

Отже, ви можете використати трейт, щоб створити класи шин команд і запитів. Наприклад, ви могли б створити окремий клас QueryBus і впроваджувати його всюди, де вам потрібна поведінка шини запитів, замість MessageBusInterface:

// src/MessageBus/QueryBus.php
namespace App\MessageBus;

use Symfony\Component\Messenger\Envelope;
use Symfony\Component\Messenger\HandleTrait;
use Symfony\Component\Messenger\MessageBusInterface;

class QueryBus
{
    use HandleTrait;

    // параметр має називатися $queryBus, щоб спрацювало автовпровадження сервісу 'query.bus'
    public function __construct(MessageBusInterface $queryBus)
    {
        // HandleTrait потребує властивості з іменем $messageBus
        $this->messageBus = $queryBus;
    }

    /**
     * @param object|Envelope $query
     *
     * @return mixed Значення, яке повернув обробник
     */
    public function query($query): mixed
    {
        return $this->handle($query);
    }
}

Під час обробки повідомлення ви також можете додати нові марки; вони будуть додані до наявних:

$this->handle(new SomeMessage($data), [new SomeStamp(), new AnotherStamp()]);

Налаштування обробників

Ручне налаштування обробників

Зазвичай Symfony знаходить і реєструє ваш обробник автоматично. Але ви можете налаштувати обробник і вручну - і передати йому додаткову конфігурацію - за допомогою атрибута #AsMessageHandler або тегуючи сервіс обробника тегом messenger.message_handler.

// src/MessageHandler/SmsNotificationHandler.php
namespace App\MessageHandler;

use App\Message\OtherSmsNotification;
use App\Message\SmsNotification;
use Symfony\Component\Messenger\Attribute\AsMessageHandler;

#[AsMessageHandler(fromTransport: 'async', priority: 10)]
class SmsNotificationHandler
{
    public function __invoke(SmsNotification $message): void
    {
        // ...
    }
}

# config/services.yaml
services:
    App\MessageHandler\SmsNotificationHandler:
        tags: [messenger.message_handler]

        # або налаштувати з опціями
        tags:
            -
                name: messenger.message_handler
                # потрібно лише тоді, коли тип неможливо вгадати за type-hint
                handles: App\Message\SmsNotification

// config/services.php
namespace Symfony\Component\DependencyInjection\Loader\Configurator;

use App\Message\SmsNotification;
use App\MessageHandler\SmsNotificationHandler;

return App::config([
    'services' => [
        SmsNotificationHandler::class => [
            'tags' => ['messenger.message_handler'],

            // або налаштувати з опціями
            'tags' => [
                [
                    'messenger.message_handler' => [
                        // потрібно лише тоді, коли тип неможливо вгадати за type-hint
                        'handles' => SmsNotification::class,
                    ],
                ],
            ],
        ],
    ],
]);

Опції, які можна налаштувати тегами:

• bus - назва шини, з якої обробник може отримувати повідомлення, за замовчуванням усі шини.
• from_transport - назва транспорту, з якого обробник може отримувати повідомлення, за замовчуванням усі транспорти.
• handles - тип повідомлень (FQCN), які може обробляти обробник; потрібно лише тоді, коли тип неможливо вгадати за type-hint.
• method - назва методу, який оброблятиме повідомлення.
• priority - визначає порядок виконання обробника, коли одне повідомлення можуть обробити кілька обробників. Обробники з вищим пріоритетом запускаються першими, і кожен наступний стартує лише після того, як попередній повністю завершився.
• sign - чи потрібно криптографічно підписувати повідомлення, які обробляє цей обробник, щоб запобігти підробці. Коли опція увімкнена, повідомлення підписуються HMAC із секретним ключем застосунку. За замовчуванням: false.

Обробка кількох повідомлень

Один клас обробника може обробляти кілька повідомлень. Для цього додайте атрибут #AsMessageHandler до всіх методів обробки:

// src/MessageHandler/SmsNotificationHandler.php
namespace App\MessageHandler;

use App\Message\OtherSmsNotification;
use App\Message\SmsNotification;

class SmsNotificationHandler
{
    #[AsMessageHandler]
    public function handleSmsNotification(SmsNotification $message): void
    {
        // ...
    }

    #[AsMessageHandler]
    public function handleOtherSmsNotification(OtherSmsNotification $message): void
    {
        // ...
    }
}

Транзакційні повідомлення: обробка нових повідомлень після завершення обробки

Обробник повідомлення може dispatch нові повідомлення під час обробки інших - у ту саму або в іншу шину (якщо застосунок має кілька шин). Будь-які помилки чи винятки, що виникають у цьому процесі, можуть мати небажані наслідки, як-от:

1. Якщо ви використовуєте DoctrineTransactionMiddleware і відправлене повідомлення викидає виняток, то всі транзакції бази даних в оригінальному обробнику буде відкочено.
2. Якщо повідомлення відправлено в іншу шину, то воно буде оброблене навіть тоді, коли код далі в поточному обробнику викине виняток.

Приклад процесу RegisterUser

Розгляньмо застосунок, що має і шину команд, і шину подій. Застосунок відправляє команду з назвою RegisterUser у шину команд. Команду обробляє RegisterUserHandler, який створює обʼєкт User, зберігає його в базу даних і відправляє повідомлення UserRegistered у шину подій.

Повідомлення UserRegistered має багато обробників, один із них може надсилати вітальний лист новому користувачеві. У цьому прикладі використано DoctrineTransactionMiddleware, щоб загорнути всі запити до бази даних в одну транзакцію.

Проблема 1: якщо під час надсилання вітального листа буде викинуто виняток, то користувача не буде створено, бо DoctrineTransactionMiddleware відкотить транзакцію Doctrine, у якій користувача було створено.

Проблема 2: якщо виняток буде викинуто під час збереження користувача в базу даних, вітальний лист усе одно надішлеться, бо він обробляється асинхронно.

Middleware DispatchAfterCurrentBusMiddleware

Для багатьох застосунків бажана поведінка - обробляти повідомлення, відправлені обробником, лише після того, як цей обробник повністю завершив роботу. Цього можна досягти, використавши DispatchAfterCurrentBusMiddleware і додавши марку DispatchAfterCurrentBusStamp до конверта повідомлення:

// src/Messenger/CommandHandler/RegisterUserHandler.php
namespace App\Messenger\CommandHandler;

use App\Entity\User;
use App\Messenger\Command\RegisterUser;
use App\Messenger\Event\UserRegistered;
use Doctrine\ORM\EntityManagerInterface;
use Symfony\Component\Messenger\Envelope;
use Symfony\Component\Messenger\MessageBusInterface;
use Symfony\Component\Messenger\Stamp\DispatchAfterCurrentBusStamp;

class RegisterUserHandler
{
    public function __construct(
        private MessageBusInterface $eventBus,
        private EntityManagerInterface $em,
    ) {
    }

    public function __invoke(RegisterUser $command): void
    {
        $user = new User($command->getUuid(), $command->getName(), $command->getEmail());
        $this->em->persist($user);

        // DispatchAfterCurrentBusStamp позначає повідомлення події так, щоб воно
        // оброблялося лише якщо цей обробник не викине виняток.

        $event = new UserRegistered($command->getUuid());
        $this->eventBus->dispatch(
            new Envelope($event)
                ->with(new DispatchAfterCurrentBusStamp())
        );

        // ...
    }
}

// src/Messenger/EventSubscriber/WhenUserRegisteredThenSendWelcomeEmail.php
namespace App\Messenger\EventSubscriber;

use App\Entity\User;
use App\Messenger\Event\UserRegistered;
use Doctrine\ORM\EntityManagerInterface;
use Symfony\Component\Mailer\MailerInterface;
use Symfony\Component\Mime\RawMessage;

class WhenUserRegisteredThenSendWelcomeEmail
{
    public function __construct(
        private MailerInterface $mailer,
        private EntityManagerInterface $em,
    ) {
    }

    public function __invoke(UserRegistered $event): void
    {
        $user = $this->em->getRepository(User::class)->find($event->getUuid());

        $this->mailer->send(new RawMessage('Welcome '.$user->getFirstName()));
    }
}

Це означає, що повідомлення UserRegistered не буде оброблене, доки RegisterUserHandler не завершиться й новий User не буде збережений у базу даних. Якщо в RegisterUserHandler виникне виняток, подія UserRegistered ніколи не буде оброблена. А якщо виняток буде викинуто під час надсилання вітального листа, транзакцію Doctrine не буде відкочено.

Примітка. Якщо WhenUserRegisteredThenSendWelcomeEmail викине виняток, цей виняток буде загорнуто в DelayedMessageHandlingException. Виклик DelayedMessageHandlingException::getWrappedExceptions дасть вам усі винятки, викинуті під час обробки повідомлення з маркою DispatchAfterCurrentBusStamp.

Middleware dispatch_after_current_bus увімкнений за замовчуванням. Якщо ви налаштовуєте middleware вручну, обовʼязково зареєструйте dispatch_after_current_bus перед doctrine_transaction у ланцюжку middleware. Крім того, middleware dispatch_after_current_bus має бути завантажений для всіх шин, які ви використовуєте.

Привʼязка обробників до різних транспортів

Кожне повідомлення може мати кілька обробників, і коли повідомлення споживається, викликаються всі його обробники. Але ви також можете налаштувати обробник так, щоб його викликали лише тоді, коли повідомлення отримано з конкретного транспорту. Це дозволяє мати одне повідомлення, кожен обробник якого викликається іншим «обробником черги», що споживає інший транспорт.

Припустімо, у вас є повідомлення UploadedImage із двома обробниками:

• ThumbnailUploadedImageHandler: ви хочете, щоб його обробляв транспорт під назвою image_transport

• NotifyAboutNewUploadedImageHandler: ви хочете, щоб його обробляв транспорт під назвою async_priority_normal

Щоб це зробити, додайте кожному обробнику опцію from_transport. Наприклад:

// src/MessageHandler/ThumbnailUploadedImageHandler.php
namespace App\MessageHandler;

use App\Message\UploadedImage;

#[AsMessageHandler(fromTransport: 'image_transport')]
class ThumbnailUploadedImageHandler
{
    public function __invoke(UploadedImage $uploadedImage): void
    {
        // створюємо мініатюру
    }
}

І аналогічно:

// src/MessageHandler/NotifyAboutNewUploadedImageHandler.php
// ...

#[AsMessageHandler(fromTransport: 'async_priority_normal')]
class NotifyAboutNewUploadedImageHandler
{
    // ...
}

Далі не забудьте «змаршрутизувати» ваше повідомлення в обидва транспорти:

# config/packages/messenger.yaml
framework:
    messenger:
        transports:
            async_priority_normal: # ...
            image_transport: # ...

        routing:
            # ...
            'App\Message\UploadedImage': [image_transport, async_priority_normal]

// config/packages/messenger.php
namespace Symfony\Component\DependencyInjection\Loader\Configurator;

return App::config([
    'framework' => [
        'messenger' => [
            'transports' => [
                'async_priority_normal' => // ...,
                'image_transport' => // ...,
            ],
            'routing' => [
                'App\Message\UploadedImage' => ['image_transport', 'async_priority_normal'],
            ],
        ],
    ],
]);

Готово! Тепер ви можете споживати кожен транспорт:

# під час обробки повідомлення викличе лише ThumbnailUploadedImageHandler
php bin/console messenger:consume image_transport -vv

php bin/console messenger:consume async_priority_normal -vv

Увага! Якщо обробник не має конфігурації from_transport, він виконуватиметься для кожного транспорту, з якого отримано повідомлення.

Обробка повідомлень пакетами

Замість обробляти повідомлення по одному, ви можете групувати їх у пакети. Обробник накопичує повідомлення, доки пакет не заповниться, а потім обробляє їх усі разом. Щоб створити пакетний обробник, реалізуйте Symfony\Component\Messenger\Handler\BatchHandlerInterface і використайте Symfony\Component\Messenger\Handler\BatchHandlerTrait:

use Symfony\Component\Messenger\Handler\Acknowledger;
use Symfony\Component\Messenger\Handler\BatchHandlerInterface;
use Symfony\Component\Messenger\Handler\BatchHandlerTrait;

class MyBatchHandler implements BatchHandlerInterface
{
    use BatchHandlerTrait;

    public function __invoke(MyMessage $message, ?Acknowledger $ack = null): mixed
    {
        return $this->handle($message, $ack);
    }

    private function process(array $jobs): void
    {
        foreach ($jobs as [$message, $ack]) {
            try {
                // обчислюємо $result з $message...

                // підтверджуємо обробку повідомлення
                $ack->ack($result);
            } catch (\Throwable $e) {
                $ack->nack($e);
            }
        }
    }
}

Коли аргумент $ack методу __invoke() дорівнює null, повідомлення має оброблятися синхронно. Інакше __invoke() має повернути кількість повідомлень, що очікують. Symfony\Component\Messenger\Handler\BatchHandlerTrait робить це за вас.

За замовчуванням пакети обробляються групами по 10 повідомлень. Щоб змінити це, перевизначте метод getBatchSize():

class MyBatchHandler implements BatchHandlerInterface
{

    // ...

    private function getBatchSize(): int
    {
        return 100;
    }
}

Пакети, що очікують, також скидаються, коли обробник черги простоює або зупиняється. Трейт визначає типовий таймаут простою в 1 секунду: якщо за цей час не надходять нові повідомлення, накопичений пакет скидається незалежно від його розміру. Щоб змінити це значення, перевизначте метод getIdleTimeout(), або поверніть null, щоб повністю вимкнути таймаут простою:

class MyBatchHandler implements BatchHandlerInterface
{

    // ...

    private function getIdleTimeout(): ?int
    {
        return 5;
    }
}

Нове у версії 8.1. Метод getIdleTimeout() було додано в Symfony 8.1.

Розширення Messenger

Конверти й марки

Повідомленням може бути будь-який PHP-обʼєкт. Іноді вам треба налаштувати щось додаткове щодо повідомлення - наприклад, спосіб його обробки всередині AMQP або затримку перед обробкою. Це робиться додаванням «марки» (stamp) до повідомлення:

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

public function index(MessageBusInterface $bus): void
{
    // чекаємо 5 секунд перед обробкою
    $bus->dispatch(new SmsNotification('...'), [
        new DelayStamp(5000),
    ]);

    // або явно створюємо Envelope
    $bus->dispatch(new Envelope(new SmsNotification('...'), [
        new DelayStamp(5000),
    ]));

    // ...
}

Усередині кожне повідомлення загортається в Envelope, який містить повідомлення й марки. Ви можете створити його вручну або дозволити це зробити шині повідомлень. Існує багато різних марок для різних цілей, і вони використовуються внутрішньо, щоб відстежувати інформацію про повідомлення - наприклад, яка шина його обробляє або чи повторюється воно після помилки.

Порада. Symfony не впроваджує Symfony\Component\Messenger\Envelope автоматично, коли ви додаєте його як аргумент методу __invoke() у вашому обробнику. Щоб цього досягти, ви можете створити такий власний middleware, який поставить марку на конверт до того, як спрацює HandleMessageMiddleware:

final class InjectEnvelopeMiddleware implements MiddlewareInterface
{
    public function handle(Envelope $envelope, StackInterface $stack): Envelope
    {
        return $stack->next()->handle(
            $envelope->with(new HandlerArgumentsStamp([$envelope])),
            $stack
        );
    }
}

Якщо вам потрібно додати до повідомлення метадані або якусь конфігурацію, загорніть його класом Symfony\Component\Messenger\Envelope і додайте марки. Наприклад, щоб задати групи серіалізації, які використовуються, коли повідомлення проходить через шар транспорту, використайте марку SerializerStamp:

use Symfony\Component\Messenger\Envelope;
use Symfony\Component\Messenger\Stamp\SerializerStamp;

$bus->dispatch(
    new Envelope($message)->with(new SerializerStamp([
        // групи застосовуються до всього повідомлення, тож не забудьте
        // визначити групу для кожного вкладеного обʼєкта
        'groups' => ['my_serialization_groups'],
    ]))
);

Ось кілька важливих марок конверта, які постачаються разом із Symfony Messenger:

• Symfony\Component\Messenger\Stamp\DelayStamp - щоб відкласти обробку асинхронного повідомлення.
• Symfony\Component\Messenger\Stamp\PriorityStamp - щоб пріоритезувати повідомлення одного типу в межах черги (доступно лише для транспортів AMQP і Beanstalkd).
• Symfony\Component\Messenger\Stamp\DispatchAfterCurrentBusStamp - щоб повідомлення оброблялося після того, як поточна шина відпрацює. Докладніше - у розділі про транзакційні повідомлення.
• Symfony\Component\Messenger\Stamp\HandledStamp - марка, що позначає повідомлення як оброблене конкретним обробником. Дає доступ до значення, яке повернув обробник, і до його імені.
• Symfony\Component\Messenger\Stamp\ReceivedStamp - внутрішня марка, що позначає повідомлення як отримане з транспорту.
• Symfony\Component\Messenger\Stamp\SentStamp - марка, що позначає повідомлення як надіслане конкретним відправником. Дає доступ до FQCN відправника та до аліаса, якщо він доступний із Symfony\Component\Messenger\Transport\Sender\SendersLocator.
• Symfony\Component\Messenger\Stamp\SerializerStamp - щоб налаштувати групи серіалізації, які використовує транспорт.
• Symfony\Component\Messenger\Stamp\ValidationStamp - щоб налаштувати групи валідації, які використовуються, коли увімкнено middleware валідації.
• Symfony\Component\Messenger\Stamp\ErrorDetailsStamp - внутрішня марка, яка додається, коли повідомлення зазнає невдачі через виняток в обробнику.
• Symfony\Component\Scheduler\Messenger\ScheduledStamp - марка, що позначає повідомлення як створене планувальником. Це допомагає відрізнити його від повідомлень, створених «вручну». Докладніше про неї можна дізнатися в документації Scheduler.

Нове у версії 8.1. PriorityStamp було додано в Symfony 8.1.

Примітка. Марка Symfony\Component\Messenger\Stamp\ErrorDetailsStamp містить Symfony\Component\ErrorHandler\Exception\FlattenException - представлення винятку, через який повідомлення зазнало невдачі. Отримати цей виняток можна методом Symfony\Component\Messenger\Stamp\ErrorDetailsStamp::getFlattenException. Виняток нормалізується завдяки Symfony\Component\Messenger\Transport\Serialization\Normalizer\FlattenExceptionNormalizer, що допомагає зі звітуванням про помилки в контексті Messenger.

Якщо ви хочете переглянути всі марки на конверті, використайте метод $envelope->all(), який повертає всі марки, згруповані за типом (FQCN). Ви також можете пройтися по всіх марках конкретного типу, передавши FQCN першим параметром цього методу (наприклад, $envelope->all(ReceivedStamp::class)).

Примітка. Будь-яка марка має бути серіалізованою компонентом Symfony Serializer, якщо вона проходить через транспорт із базовим серіалізатором Symfony\Component\Messenger\Transport\Serialization\Serializer.

Типові марки на повідомленнях

Повідомлення можуть визначати власні типові марки для відправлення, реалізувавши Symfony\Component\Messenger\Message\DefaultStampsProviderInterface.

Це корисно, коли повідомлення завжди потребує тих самих марок, як-от марки затримки чи марки маршрутизації. Замість передавати ці марки в кожному місці відправлення, саме повідомлення може оголосити їх один раз:

use Symfony\Component\Messenger\Message\DefaultStampsProviderInterface;
use Symfony\Component\Messenger\Stamp\DelayStamp;

class ExportReportMessage implements DefaultStampsProviderInterface
{
    public function getDefaultStamps(): array
    {
        // цю марку буде додано автоматично під час відправлення
        return [new DelayStamp(1000)];
    }
}

Під час відправлення повідомлення типові марки додаються лише тоді, коли на конверті ще немає іншої марки того самого класу:

// DelayStamp(1000) додається за замовчуванням
$bus->dispatch(new ExportReportMessage());

// явна DelayStamp(500) перекриває типову
$bus->dispatch(
    new ExportReportMessage(),
    [new DelayStamp(500)]
);

Middleware

Те, що відбувається, коли ви відправляєте повідомлення в шину повідомлень, залежить від набору її middleware та їхнього порядку. За замовчуванням middleware, налаштовані для кожної шини, такі:

1. add_bus_name_stamp_middleware - додає марку, яка фіксує, у яку шину було відправлено це повідомлення;

2. dispatch_after_current_bus - див. розділ про транзакційні повідомлення;

3. failed_message_processing_middleware - обробляє повідомлення, які повторюються через транспорт помилок (failure transport), щоб вони працювали належно, ніби їх отримано з їхнього оригінального транспорту;

4. decode_failed_message - коли повідомлення неможливо декодувати (наприклад, його клас було видалено), цей middleware намагається декодувати його заново за допомогою серіалізатора транспорту, даючи змогу відновити повідомлення після того, як виправлення буде розгорнуто;

   Нове у версії 8.1. Middleware decode_failed_message було додано в Symfony 8.1.

5. Ваш власний набір middleware;

6. send_message - якщо для транспорту налаштована маршрутизація, надсилає повідомлення в цей транспорт і зупиняє ланцюжок middleware;

7. handle_message - викликає обробник (чи обробники) повідомлення для заданого повідомлення.

Примітка. Ці назви middleware насправді є скороченнями. Справжні ідентифікатори сервісів мають префікс messenger.middleware. (наприклад, messenger.middleware.handle_message).

Middleware виконуються під час відправлення повідомлення, а також ще раз, коли повідомлення отримує обробник черги (для повідомлень, надісланих у транспорт для асинхронної обробки). Памʼятайте про це, коли пишете власний middleware.

Ви можете додати власні middleware до цього списку або повністю вимкнути стандартні й залишити лише свої.

Якщо сервіс middleware абстрактний, ви можете налаштувати аргументи його конструктора, і для кожної шини буде створено окремий екземпляр.

# config/packages/messenger.yaml
framework:
    messenger:
        buses:
            messenger.bus.default:
                # вимикаємо стандартні middleware
                default_middleware: false

                middleware:
                    # використовуємо й налаштовуємо ті частини стандартних middleware, які потрібні
                    - 'add_bus_name_stamp_middleware': ['messenger.bus.default']

                    # додаємо власні сервіси, що реалізують Symfony\Component\Messenger\Middleware\MiddlewareInterface
                    - 'App\Middleware\MyMiddleware'
                    - 'App\Middleware\AnotherMiddleware'

// config/packages/messenger.php
namespace Symfony\Component\DependencyInjection\Loader\Configurator;

return App::config([
    'framework' => [
        'messenger' => [
            'buses' => [
                'messenger.bus.default' => [
                    'default_middleware' => false,
                    'middleware' => [
                        // використовуємо й налаштовуємо ті частини стандартних middleware, які потрібні
                        'add_bus_name_stamp_middleware' => ['messenger.bus.default'],

                        // додаємо власні сервіси, що реалізують Symfony\Component\Messenger\Middleware\MiddlewareInterface
                        'App\Middleware\MyMiddleware',
                        'App\Middleware\AnotherMiddleware',
                    ],
                ],
            ],
        ],
    ],
]);

Порада: якщо у вас встановлено MakerBundle, скористайтеся командою make:messenger-middleware, щоб згенерувати заготовку власного messenger middleware.

Middleware має реалізовувати Symfony\Component\Messenger\Middleware\MiddlewareInterface. Замість того щоб працювати безпосередньо з повідомленнями, middleware отримує конверт (envelope). Тож він може переглянути вміст конверта і його марки (stamps) або додати власні:

use App\Message\Stamp\AnotherStamp;
use Symfony\Component\Messenger\Envelope;
use Symfony\Component\Messenger\Middleware\MiddlewareInterface;
use Symfony\Component\Messenger\Middleware\StackInterface;
use Symfony\Component\Messenger\Stamp\ReceivedStamp;

class MyOwnMiddleware implements MiddlewareInterface
{
    public function handle(Envelope $envelope, StackInterface $stack): Envelope
    {
        if (null !== $envelope->last(ReceivedStamp::class)) {
            // Повідомлення щойно отримано...

            // Можна, наприклад, додати ще одну марку.
            $envelope = $envelope->with(new AnotherStamp(/* ... */));
        } else {
            // Повідомлення щойно було відправлено
        }

        return $stack->next()->handle($envelope, $stack);
    }
}

Наведений приклад передає повідомлення наступному middleware з додатковою маркою, якщо повідомлення щойно було отримано (тобто має принаймні одну марку ReceivedStamp). Власні марки створюються реалізацією Symfony\Component\Messenger\Stamp\StampInterface.

Дедуплікація повідомлень

Symfony має middleware, який за допомогою блокувань не дає відправити або обробити одне й те саме повідомлення кілька разів.

Ця поведінка вмикається додаванням марки Symfony\Component\Messenger\Stamp\DeduplicateStamp до конверта повідомлення. Middleware використовує key з марки, щоб визначити, чи потрібно пропустити повідомлення:

use Symfony\Component\Messenger\Stamp\DeduplicateStamp;

$message = new MyMessage($projectId);

// не даємо одночасно обробляти кілька повідомлень для одного проєкту
$deduplicationKey = 'my_message.project.'.$projectId;

$bus->dispatch($message, [
    new DeduplicateStamp($deduplicationKey),
]);

// другий аргумент задає TTL блокування (за замовчуванням 300 секунд)
new DeduplicateStamp($deduplicationKey, 3600),

Ключ дедуплікації - це рішення рівня бізнес-логіки: він кодує, що саме для вашого застосунку означає «те саме повідомлення». Глобальна унікальність не потрібна, але ключ має визначати, коли два повідомлення слід вважати дублікатами (наприклад, за ID проєкту, ID замовлення або комбінацією потрібних полів).

Типово дедуплікація діє, поки повідомлення перебуває в черзі й поки воно обробляється. Блокування знімається після успішної обробки. Якщо дедуплікація потрібна лише на час перебування в черзі, передайте третім аргументом true:

new DeduplicateStamp($deduplicationKey, 300, true)

У цьому режимі блокування знімається, щойно обробник черги отримує повідомлення, тож інше повідомлення з тим самим ключем може оброблятися паралельно.

Коли обробник кидає виняток і повідомлення йде на повторну спробу, блокування зберігається, щоб повторні спроби дедуплікувалися відносно нових відправлень із тим самим ключем. Коли цикл повторних спроб вичерпано, блокування знімається, зокрема й тоді, коли повідомлення переміщується у транспорт для невдалих повідомлень.

Новинка у версії 8.1: зняття блокування дедуплікації при остаточній невдачі зʼявилося в Symfony 8.1.

Зверніть увагу: цей middleware вмикається автоматично, якщо встановлено компонент Lock.

Middleware для Doctrine

Якщо у вашому застосунку використовується Doctrine, доступні кілька необовʼязкових middleware, які можуть стати в пригоді:

# config/packages/messenger.yaml
framework:
    messenger:
        buses:
            command_bus:
                middleware:
                    # щоразу під час обробки повідомлення зʼєднання Doctrine
                    # "пінгується" і перевідкривається, якщо воно закрите. Корисно,
                    # якщо ваші обробники черги працюють довго, а зʼєднання з базою
                    # даних час від часу втрачається
                    - doctrine_ping_connection

                    # після обробки зʼєднання Doctrine закривається,
                    # що звільняє зʼєднання з базою в обробнику черги,
                    # замість тримати їх відкритими вічно
                    - doctrine_close_connection

                    # записує помилку в лог, якщо транзакцію Doctrine відкрито, але не закрито
                    - doctrine_open_transaction_logger

                    # обгортає всі обробники в одну транзакцію Doctrine
                    # обробникам не потрібно викликати flush(), а помилка
                    # в будь-якому з них спричинить відкат
                    - doctrine_transaction

                    # або передайте будь-якому з них інший entity manager
                    #- doctrine_transaction: ['custom']

// config/packages/messenger.php
namespace Symfony\Component\DependencyInjection\Loader\Configurator;

return App::config([
    'framework' => [
        'messenger' => [
            'buses' => [
                'command_bus' => [
                    'middleware' => [
                        // щоразу під час обробки повідомлення зʼєднання Doctrine
                        // "пінгується" і перевідкривається, якщо воно закрите. Корисно,
                        // якщо ваші обробники черги працюють довго, а зʼєднання з базою
                        // даних час від часу втрачається
                        'doctrine_ping_connection',

                        // після обробки зʼєднання Doctrine закривається,
                        // що звільняє зʼєднання з базою в обробнику черги,
                        // замість тримати їх відкритими вічно
                        'doctrine_close_connection',

                        // записує помилку в лог, якщо транзакцію Doctrine відкрито, але не закрито
                        'doctrine_open_transaction_logger',

                        // обгортає всі обробники в одну транзакцію Doctrine
                        // обробникам не потрібно викликати flush(), а помилка
                        // в будь-якому з них спричинить відкат
                        'doctrine_transaction',

                        // або передайте будь-якому з них інший entity manager
                        // 'doctrine_transaction' => ['custom'],
                    ],
                ],
            ],
        ],
    ],
]);

Інші middleware

Додайте middleware router_context, якщо вам потрібно генерувати абсолютні URL у споживачі (наприклад, рендерити шаблон із посиланнями). Цей middleware зберігає контекст початкового запиту (хост, HTTP-порт тощо), який потрібен для побудови абсолютних URL.

Додайте middleware validation, якщо перед обробкою потрібно валідувати обʼєкт повідомлення за допомогою компонента Validator. Якщо валідація не пройде, буде кинуто ValidationFailedException. Групи валідації налаштовуються маркою Symfony\Component\Messenger\Stamp\ValidationStamp.

# config/packages/messenger.yaml
framework:
    messenger:
        buses:
            command_bus:
                middleware:
                    - router_context
                    - validation

// config/packages/messenger.php
namespace Symfony\Component\DependencyInjection\Loader\Configurator;

return App::config([
    'framework' => [
        'messenger' => [
            'buses' => [
                'command_bus' => [
                    'middleware' => [
                        'router_context',
                        'validation',
                    ],
                ],
            ],
        ],
    ],
]);

Події Messenger

Окрім middleware, Messenger відправляє кілька подій. Ви можете створити слухач подій, щоб підключитися до різних етапів процесу. Для кожної з них імʼям події є клас події:

• Symfony\Component\Messenger\Event\SendMessageToTransportsEvent
• Symfony\Component\Messenger\Event\MessageSentToTransportsEvent
• Symfony\Component\Messenger\Event\WorkerMessageFailedEvent
• Symfony\Component\Messenger\Event\WorkerMessageHandledEvent
• Symfony\Component\Messenger\Event\WorkerMessageReceivedEvent
• Symfony\Component\Messenger\Event\WorkerMessageRetriedEvent
• Symfony\Component\Messenger\Event\WorkerRateLimitedEvent
• Symfony\Component\Messenger\Event\WorkerRunningEvent
• Symfony\Component\Messenger\Event\WorkerStartedEvent
• Symfony\Component\Messenger\Event\WorkerStoppedEvent

Зверніть увагу: подія MessageSentToTransportsEvent відправляється лише після того, як повідомлення надіслано щонайменше в один транспорт. Якщо повідомлення надіслано в кілька транспортів, подія відправляється лише один раз.

Додаткові аргументи обробника

Messenger може передати обробнику повідомлення додаткові дані за допомогою Symfony\Component\Messenger\Stamp\HandlerArgumentsStamp. Додайте цю марку до конверта в middleware й заповніть її будь-якими даними, які мають бути доступні в обробнику:

// src/Messenger/AdditionalArgumentMiddleware.php
namespace App\Messenger;

use Symfony\Component\Messenger\Envelope;
use Symfony\Component\Messenger\Middleware\MiddlewareInterface;
use Symfony\Component\Messenger\Middleware\StackInterface;
use Symfony\Component\Messenger\Stamp\HandlerArgumentsStamp;

final class AdditionalArgumentMiddleware implements MiddlewareInterface
{
    public function handle(Envelope $envelope, StackInterface $stack): Envelope
    {
        $envelope = $envelope->with(new HandlerArgumentsStamp([
            $this->resolveAdditionalArgument($envelope->getMessage()),
        ]));

        return $stack->next()->handle($envelope, $stack);
    }

    private function resolveAdditionalArgument(object $message): mixed
    {
        // ...
    }
}

Тоді ваш обробник матиме такий вигляд:

// src/MessageHandler/SmsNotificationHandler.php
namespace App\MessageHandler;

use App\Message\SmsNotification;

final class SmsNotificationHandler
{
    public function __invoke(SmsNotification $message, mixed $additionalArgument)
    {
        // ...
    }
}

Серіалізатор повідомлень для власних форматів даних

Якщо ви отримуєте повідомлення від інших застосунків, вони можуть мати не той формат, який вам потрібен. Не кожен застосунок повертає JSON-повідомлення з полями body і headers. У таких випадках доведеться створити новий серіалізатор повідомлень, який реалізує Symfony\Component\Messenger\Transport\Serialization\SerializerInterface. Припустімо, вам потрібен декодер повідомлень:

namespace App\Messenger\Serializer;

use Symfony\Component\Messenger\Envelope;
use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;

class MessageWithTokenDecoder implements SerializerInterface
{
    public function decode(array $encodedEnvelope): Envelope
    {
        try {
            // розбираємо отримані дані з вашими власними полями
            $data = $encodedEnvelope['data'];
            $data['token'] = $encodedEnvelope['token'];

            // інші операції, наприклад отримання інформації з марок
        } catch (\Throwable $throwable) {
            // обгортаємо будь-який виняток у конверт, щоб надіслати його у транспорт для невдалих повідомлень
            return new Envelope($throwable);
        }

        return new Envelope($data);
    }

    public function encode(Envelope $envelope): array
    {
        // цей декодер не кодує повідомлення, але ви можете це реалізувати, повернувши
        // масив із серіалізованими марками, якщо потрібно надсилати повідомлення у власному форматі
        throw new \LogicException('This serializer is only used for decoding messages.');
    }
}

Наступний крок - вказати Symfony використовувати цей серіалізатор в одному або кількох ваших транспортах:

# config/packages/messenger.yaml
framework:
    messenger:
        transports:
            my_transport:
                dsn: '%env(MY_TRANSPORT_DSN)%'
                serializer: 'App\Messenger\Serializer\MessageWithTokenDecoder'

// config/packages/messenger.php
namespace Symfony\Component\DependencyInjection\Loader\Configurator;

use App\Messenger\Serializer\MessageWithTokenDecoder;

return App::config([
    'framework' => [
        'messenger' => [
            'transports' => [
                'my_transport' => [
                    'dsn' => env('MY_TRANSPORT_DSN'),
                    'serializer' => MessageWithTokenDecoder::class,
                ],
            ],
        ],
    ],
]);

Кілька шин: командна шина і шина подій

Типово Messenger дає вам один сервіс шини повідомлень. Але ви можете налаштувати скільки завгодно шин, створивши «командну», «запитну» чи «подієву» шину й керуючи їхніми middleware.

Поширена архітектура під час побудови застосунків - відокремлювати команди від запитів. Команди щось роблять, а запити отримують дані. Це називають CQRS (Command Query Responsibility Segregation). Детальніше про це - у статті Мартіна Фаулера про CQRS. Таку архітектуру можна поєднати з компонентом Messenger, визначивши кілька шин.

Командна шина дещо відрізняється від запитної шини. Наприклад, командні шини зазвичай не повертають результатів, а запитні рідко бувають асинхронними. Правила цих шин налаштовуються через middleware.

Непоганою ідеєю буває також відокремити дії від реакцій, увівши шину подій. Шина подій може мати нуль або більше підписників.

Порада: одна шина - хороший вибір за замовчуванням. Додавайте другу шину лише тоді, коли вам потрібен інший набір middleware, а не тому, що так радить архітектурний патерн. Наприклад, ви можете додати middleware doctrine_transaction до командної шини й не додавати його до запитної, бо операціям читання транзакція ні до чого.

Якщо ви не можете назвати конкретну різницю в поведінці між шинами, вам, найпевніше, поки що не потрібно більше однієї.

framework:
    messenger:
        # Шина, яку буде впроваджено при впровадженні MessageBusInterface
        default_bus: command.bus
        buses:
            command.bus:
                middleware:
                    - validation
                    - doctrine_transaction
            query.bus:
                middleware:
                    - validation
            event.bus:
                default_middleware:
                    enabled: true
                    # встановіть "allow_no_handlers" у true (типово false), щоб дозволити
                    # відсутність налаштованого обробника для цієї шини без винятку
                    allow_no_handlers: false
                    # встановіть "allow_no_senders" у false (типово true), щоб кидати виняток,
                    # якщо для цієї шини не налаштовано жодного відправника
                    allow_no_senders: true
                middleware:
                    - validation

// config/packages/messenger.php
namespace Symfony\Component\DependencyInjection\Loader\Configurator;

return App::config([
    'framework' => [
        'messenger' => [
            // Шина, яку буде впроваджено при впровадженні MessageBusInterface
            'default_bus' => 'command.bus',
            'buses' => [
                'command.bus' => [
                    'middleware' => [
                        'validation',
                        'doctrine_transaction',
                    ],
                ],
                'query.bus' => [
                    'middleware' => [
                        'validation',
                    ],
                ],
                'event.bus' => [
                    'default_middleware' => [
                        'enabled' => true,
                        // встановіть "allow_no_handlers" у true (типово false), щоб дозволити
                        // відсутність налаштованого обробника для цієї шини без винятку
                        'allow_no_handlers' => false,
                        // встановіть "allow_no_senders" у true (типово false), щоб дозволити
                        // відсутність налаштованого відправника для цієї шини без винятку
                        'allow_no_senders' => true,
                    ],
                    'middleware' => [
                        'validation',
                    ],
                ],
            ],
        ],
    ],
]);

Це створить три нові сервіси:

• command.bus: доступний для автоматичного впровадження за тайп-хінтом Symfony\Component\Messenger\MessageBusInterface (бо це default_bus);

• query.bus: доступний для автоматичного впровадження як MessageBusInterface $queryBus;

• event.bus: доступний для автоматичного впровадження як MessageBusInterface $eventBus.

Обмеження обробників за шиною

Типово кожен обробник доступний для обробки повідомлень на всіх ваших шинах. Щоб повідомлення не потрапило в неправильну шину без жодної помилки, обмежте кожен обробник конкретною шиною за допомогою тега messenger.message_handler:

# config/services.yaml
services:
    App\MessageHandler\SomeCommandHandler:
        tags: [{ name: messenger.message_handler, bus: command.bus }]

// config/services.php
namespace Symfony\Component\DependencyInjection\Loader\Configurator;

use App\MessageHandler\SomeCommandHandler;

return App::config([
    'services' => [
        SomeCommandHandler::class => [
            'tags' => [
                ['messenger.message_handler' => ['bus' => 'command.bus']],
            ],
        ],
    ],
]);

Так обробник App\MessageHandler\SomeCommandHandler буде відомий лише шині command.bus.

Цей тег можна автоматично додати цілій групі класів через конфігурацію сервісів _instanceof. У такий спосіб шину повідомлень можна визначити за реалізованим інтерфейсом:

# config/services.yaml
services:
    # ...

    _instanceof:
        # усі сервіси, що реалізують CommandHandlerInterface,
        # буде зареєстровано на шині command.bus
        App\MessageHandler\CommandHandlerInterface:
            tags:
                - { name: messenger.message_handler, bus: command.bus }

        # тоді як ті, що реалізують QueryHandlerInterface,
        # буде зареєстровано на шині query.bus
        App\MessageHandler\QueryHandlerInterface:
            tags:
                - { name: messenger.message_handler, bus: query.bus }

// config/services.php
namespace Symfony\Component\DependencyInjection\Loader\Configurator;

use App\MessageHandler\CommandHandlerInterface;
use App\MessageHandler\QueryHandlerInterface;

return App::config([
    'services' => [
        '_instanceof' => [
            // усі сервіси, що реалізують CommandHandlerInterface,
            // буде зареєстровано на шині command.bus
            CommandHandlerInterface::class => [
                'tags' => [
                    ['messenger.message_handler' => ['bus' => 'command.bus']],
                ],
            ],
            // тоді як ті, що реалізують QueryHandlerInterface,
            // буде зареєстровано на шині query.bus
            QueryHandlerInterface::class => [
                'tags' => [
                    ['messenger.message_handler' => ['bus' => 'query.bus']],
                ],
            ],
        ],
    ],
]);

Налагодження шин

Команда debug:messenger виводить доступні повідомлення й обробники для кожної шини. Список можна обмежити конкретною шиною, передавши її імʼя аргументом.

$ php bin/console debug:messenger

  Messenger
  =========

  command.bus
  -----------

   The following messages can be dispatched:

   ---------------------------------------------------------------------------------------
    App\Message\DummyCommand
        handled by App\MessageHandler\DummyCommandHandler
    App\Message\MultipleBusesMessage
        handled by App\MessageHandler\MultipleBusesMessageHandler
   ---------------------------------------------------------------------------------------

  query.bus
  ---------

   The following messages can be dispatched:

   ---------------------------------------------------------------------------------------
    App\Message\DummyQuery
        handled by App\MessageHandler\DummyQueryHandler
    App\Message\MultipleBusesMessage
        handled by App\MessageHandler\MultipleBusesMessageHandler
   ---------------------------------------------------------------------------------------

Порада: команда також покаже PHPDoc-опис класів повідомлення й обробника.

Повторне відправлення повідомлення

Якщо потрібно відправити повідомлення повторно (з тим самим транспортом і конвертом), створіть новий Symfony\Component\Messenger\Message\RedispatchMessage і відправте його через свою шину. Скористаємося тим самим прикладом SmsNotification, що й раніше:

// src/MessageHandler/SmsNotificationHandler.php
namespace App\MessageHandler;

use App\Message\SmsNotification;
use Symfony\Component\Messenger\Attribute\AsMessageHandler;
use Symfony\Component\Messenger\Message\RedispatchMessage;
use Symfony\Component\Messenger\MessageBusInterface;

#[AsMessageHandler]
class SmsNotificationHandler
{
    public function __construct(private MessageBusInterface $bus)
    {
    }

    public function __invoke(SmsNotification $message): void
    {
        // щось робимо з повідомленням,
        // а потім відправляємо його повторно за власною логікою

        if ($needsRedispatch) {
            $this->bus->dispatch(new RedispatchMessage($message));
        }
    }
}

Вбудований Symfony\Component\Messenger\Handler\RedispatchMessageHandler подбає про це повідомлення й відправить його повторно через ту саму шину, якою воно було відправлене спочатку. Другим аргументом конструктора RedispatchMessage можна вказати транспорти, які використовувати для повторного відправлення.

Дізнатися більше

Решта статей розділу: /messenger/*.