Возникла как то потребность использовать асинхронную отправку писем.

Передо мной открылись два основных направления:

• Наколхозить на скорую руку свою поделку для работы с очередью сообщений.

• Использовать мощные стабильные инструменты.

Выбор пал на symfony/messenger по нескольким причинам:

• Во первых, первым путём я уже хаживал.

• Во вторых, я давно смотрел в сторону Symfony ожидая подходящей идеи для личного проекта, чтобы неспешно войти в его мир.

• В третьих, он поддерживает несколько различных транспортов.

• В четвёртых, предоставляет возможность использовать практически готовые к работе воркеры, предоставляемыe компонентом symfony/console.

• Ну и последнее, впереди были выходные, и я мог себе позволить разобраться как это сделать.

Надо отметить, что официальная документация пролила мало света на то, как использовать компонент symfony/messenger без использования самого фреймворка. Все приводимые примеры рассматривают как сконфигурировать зависимости для symfony контейнера в yaml файлах.

В документации к компоненту есть статья: https://symfony.com/doc/current/components/messenger.html про использование мессенджера в качестве независимого компонента, но про настройку транспорта к сожалению в ней ничего не сказано.

Так же на Хабре есть статья - Перевод PHP бэкенда на шину Redis streams и выбор независимой от фреймворков библиотеки. В ней автор провёл важные изыскания, показал способ настройки транспорта, но по какой то причине отказался использовать symfony/console.

Из комментариев к той статье, я понял, что, существует пробел в понимании этого мощного и важного инструмента.

В статье я опущу описание создания классов сообщения и его обработчика так как в документации этот момент подробно описан.

И так приступим.

Конфигурируем соединение в контейнере используя DSN, для RabbitMQ это может выглядеть так:

AmqpConnection::class => function (ContainerInterface $container) {
    return AmqpConnection::fromDsn('amqp://guest:guest@rabbitmq:5672/%2f/messages', []);
}

для Redis так:

RedisConnection::class => function (ContainerInterface $container) {
    return RedisConnection::fromDsn('redis://redis:6379/messages', [
        'stream' => 'messages',
        'group' => 'default',
        'consumer' => 'default',
    ]);
}

Далее настраиваем рессиверы для обеих соединений, они нам понадобятся в консоли. Обратите внимание, что делаем их именованными, имменно по имени ampq-async или redis-async команда symfony/console messenger:consume ampq-async сможет его обнаружить в контейнере.

'ampq-async' => function (ContainerInterface $container) {
    return new AmqpReceiver(
        $container->get(AmqpConnection::class)
    );
},

'redis-async' => function (ContainerInterface $container) {
    return new RedisReceiver(
        $container->get(RedisConnection::class)
    );
},

Далее описываем в контейнере шину сообщений, передавая создаваемые там же SendMessageMiddleware и HandleMessageMiddleware

'message-bus' => function (ContainerInterface $container) {
    $handler = new EmailMessageHandler($container);
    return new MessageBus([ 
        new SendMessageMiddleware(
            new SendersLocator([ 
                EmailMessage::class => [
                    AmqpTransport::class,
                /*    RedisTransport::class, */
                ]
            ], $container)
        ),
        new HandleMessageMiddleware(
            new HandlersLocator([
                EmailMessage::class => [$handler],
            ])
        )
    ]);
}    

Шину так же делаем именованной, имя шины опять таки потребуется нам в консольной команде messenger:consume, но к этому ещё вернёмся позже, когда будем рассматривать работу консольного приложения.

Таким образом, получаем полностью сконфигурированную шину сообщений. Очень просто, не правда ли?

Вот и всё, теперь в контроллере отправить сообщение на шину можно просто вызвав:

$busName = 'message-bus';
$bus = $this->container->get($busName);
$bus->dispatch(
    new EmailMessage($message), [
        new Symfony\Component\Messenger\Stamp\BusNameStamp($busName) 
    ]
);

Обратите внимание, что помимо сообщения мы передаём в шину штамп с названием шины, без этого штампа нельзя использовать стандартную команду ConsumeMessagesCommand т.к. по имени в этом штампе RoutableMessageBus находит нужную шину в контейнере.

Теперь о потреблении наших сообщений из очереди.

Компонент symfony/console это очень мощный и гибкий инструмент. При использовании его в рамках фреймворка приложение использует Symfony\Bundle\FrameworkBundle\Console\Application, передавая ему конфигурацию фреймворка, таким образом консоль получает возможность использовать все команды всех доступных компонентов фреймворка.

Но нам такой радости не вкусить. В случае использования symfony/console в качестве независимого компонента, приложению придётся в качестве ядра использовать Symfony\Component\Console\Application, а все команды конфигурировать в ручную.

Установим компонент консоли.

composer require symfony/console

После установки документация предлагает создать файл console в папке bin с вот таким содержимым.

#!/usr/bin/env php
<?php
// application.php
require __DIR__.'/vendor/autoload.php';
use Symfony\Component\Console\Application;
$application = new Application();
// ... register commands
$application->run();

Теперь обустроем это файл в соответствии с нашими потребностями.

В первую очередь добавим определение нашего контейнера.

$containerBuilder = new ContainerBuilder();
$containerBuilder->addDefinitions(__DIR__ . '/../config/dependencies.php');
$container = $containerBuilder->build();

Далее нам потребуется объект консольного вывода

$output = new ConsoleOutput();

ну и конечно же определения необходимых команд

$commands = [
    new ConsumeMessagesCommand(
        new RoutableMessageBus($container),
        $container,
        new EventDispatcher(),
        new ConsoleLogger($output, [])
    ),
    new StopWorkersCommand(
        new FilesystemAdapter('', 10, __DIR__ . '/../var/cache')
    )
];    

Тут требуются некоторые пояснения.

Конструктор класса ConsumeMessagesCommand требует RoutableMessageBus, в который нужно передать сконфигурированный контейнер.

В этом контейнере он сможет найти шину по имени 'message-bus', которое мы указали в определении шины ранее и передали в штампе сообщения.

Так же нужно передать сам контейнер, EventDispatcher и ConsoleLogger с созданным ранее $output для того,

что бы иметь возможность пользоваться всеми преимуществами стандартного консольного вывода от symfony, такими как:

• расцветка вывода в зависимости от важности события

• возможность регулировать детализацию вывода стандартными ключами при запуске -v -vv -vvv.

Обратите внимание, что $output с которым мы создавали ConsoleLogger впоследствии необходимо передать вызову $application->run(null, $output);

Для команды stopWorkersCommand нужно передать адаптер кэширования, для возможности мягкой остановки воркеров, чтобы избежать ситуации когда воркер уже взял сообщение из очереди, но ещё не успел его обработать.

#!/usr/bin/env php
<?php

require __DIR__.'/../vendor/autoload.php';

use DI\ContainerBuilder;
use Symfony\Component\Messenger\Command\ConsumeMessagesCommand;
use Symfony\Component\Messenger\Command\StopWorkersCommand;
use Symfony\Component\EventDispatcher\EventDispatcher;
use Symfony\Component\Cache\Adapter\FilesystemAdapter;
use Symfony\Component\Messenger\RoutableMessageBus;
use Symfony\Component\Console\Logger\ConsoleLogger;
use Symfony\Component\Console\Output\ConsoleOutput;
use Symfony\Component\Console\Application;

$containerBuilder = new ContainerBuilder();
$containerBuilder->addDefinitions(__DIR__ . '/../config/dependencies.php');
$container = $containerBuilder->build();

$output = new ConsoleOutput();

$commands = [
    new ConsumeMessagesCommand(
        new RoutableMessageBus($container),
        $container,
        new EventDispatcher(),
        new ConsoleLogger($output, [])
    ),
    new StopWorkersCommand(
        new FilesystemAdapter('', 10, __DIR__ . '/../var/cache')
    )
];    

$application = new Application('Console');
$application->addCommands($commands);
$application->run(null, $output);

Теперь запустить консольное приложение можно так: php bin/console messenger:consume ampq-async или для redis-транспорта: php bin/console messenger:consume redis-async, и при помощи ключей -v, -vv, или -vvv управлять детализацией вывода сообщений в консоль, также остановить воркеры командой php bin/console messenger:stop-workers.

Заключение

Исследовав компоненты messenger и console, я высоко оценил удобство этих инструментов.

Для меня это был не только отличный повод разобраться в устройстве Symfony, но и статья (в хорошем смысле этого слова), ну ладно не статья, рецепт или заметочка.