Мессенджер: работа с синхронизированными сообщениями и сообщениями в очереди¶

Маршрутизация сообщений к транспорту¶

Теперь, когда у вас есть сконфигурированный транспорт, вместо немедленно обработки сообщений, вы можете сконфигурировать их так, чтобы они были отправлены транспорту:

Благодаря этому, App\Message\SmsNotification будет отправлен транспорту async, и его обработчик(и) не будут вызваны сразу же. Все сообщения, не совпавшие с routing будут обработаны немедленно.

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

Сущности Doctrine в сообщениях¶

Если вам нужно передать сущность в Doctrine в сообщении, лучше всего передать основной ключ сущности (или любую релевантную информацию, необходимую обработчику, вроде email, и др.) вместо объекта:

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

class NewUserWelcomeEmail
{
    private $userId;

    public function __construct(int $userId)
    {
        $this->userId = $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\Handler\MessageHandlerInterface;

class NewUserWelcomeEmailHandler implements MessageHandlerInterface
{
    private $userRepository;

    public function __construct(UserRepository $userRepository)
    {
        $this->userRepository = $userRepository;
    }

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

        // ... отправить электронное письмо!
    }
}

Это гарантирует, что сущность содержит свежие данные.

Сихнронная обработка сообщений¶

Если сообщение не совпадает ни с одним правилом маршрутизации, оно не будет отправлено ни одному транспорту, и будет обработано немедленно. В некоторых случаях (например, при `связывании обработчиков с разными транспортами`_), легче и более гибко обработать их ясно: создав транспорт sync и “отправив” сообщения в него для немедленной обработки:

Создание вашего собственного транспорта¶

Вы также можете создать собственный транспорт, если вам нужно отправлять или получать сообщения из чего-то, что не поддерживается. См. How to Create Your own Messenger Transport.

Запуск в производство¶

В производстве, есть несколько важных вещей, о которых стоит подумать:

Используйте Супервизора, чтобы ваш(и) работник(и) работали

Вы хотите, чтобы один или более “работников” работали все время. Чтобы сделать это, используйте систему контроля процесса вроде Супервизора.

Не позволяйте работникам работать бесконечно

Некоторые сервисы (вроде EntityManager Doctrine) будут потреблять все больше памяти со временем. Поэтому, вместо того, чтобы позволять вашему работнику работать всегда, используйте флажок вроде messenger:consume --limit=10, чтобы указать работнику, что он должен обработать только 10 сообщений до прекращения работы (затем Супервизор создаст новый процесс). Также есть другие опции вроде --memory-limit=128M и --time-limit=3600.

Перезагружайте работников при запуске

Каждый раз при запуске вам нужно будет перезагрузить все процессы ваших работников, чтобы они видели новозапущенный код. Чтобы сделать это, выполните messenger:stop-workers при запуске. Это сигнализирует каждому работнику, что он должен закончить обработку текущего сообщения и грациозно завершить работу. Затем, Супервизор создаст новые процессы работников. Команда использует кеш app внутренне - поэтому убедитесь в том, что он сконфигурирован для использования адаптера по вашему вкусу.

Используйте один кеш между запусками

Если ваша стратегия запуска заключается в создании новых целевых каталогов, вам нужно установить значение опции конфигурации cache.prefix.seed, чтобы использовать одно и то же пространство имен кеша между запусками. Иначе, пул cache.app будет использовать значение параметра kernel.project_dir в качестве базы для пространства имен, что приведет к разным пространствам имен каждый раз при запуске.

Приоритизированный транспорт¶

Иногда определенные типы сообщений должны иметь более высокий приоритет и быть обработаны до других. Чтобы сделать это возможным, вы можете создать несколько транспортов и маршрутизировать разные сообщения к ним. Например:

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

1

$ php bin/console messenger:consume async_priority_high async_priority_low

Работник будет всегда вначале искать сообщения, ожидающие async_priority_high. Если таких нет, затем он будет потреблять сообщения из async_priority_low.

Ограничьте потребление для конкретных очередей¶

Некоторый транспорт (особенно AMQP) имеет концепцию обмена и очередей. Транспорт Symfony всегда связан с обменом. По умолчанию, работник потребляет из всех очередей, соединенных с обменом указанного транспорта. Однако, есть примеры использования, когда вам нужно, чтобы работник потреблял только из конкретной очереди.

You can limit the worker to only process messages from specific queues:

1

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

Чтобы позволить использование опции queues, получатель должен реализовывать QueueReceiverInterface.

Конфигурация супервизора¶

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

1

$ sudo apt-get install supervisor

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

1
2
3
4
5
6
7
8
9

;/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
process_name=%(program_name)s_%(process_num)02d

Измените аргумент async, чтобы он использовал название вашего транспорта (или транспортов) и user на Unix-пользователя на вашем сервере.

Если вы используете транспорт Redis, заметьте, что каждому работнику нужно уникальное имя потребителя, чтобы избежать обработки одного сообщения несколькими работникам. Один из способов достичь этого - установить переменную окружения в файле конфигурации Супервизора, на которую вы потом можете сослаться в messenger.yaml (см. раздел Redis выше):

1

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

Next, tell Supervisor to read your config and start your workers:

1
2
3
4
5

$ sudo supervisorctl reread

$ sudo supervisorctl update

$ sudo supervisorctl start messenger-consume:*

См. документацию Супервизора, чтобы узнать больше деталей.

Грациозное выключение¶

Если вы установили в своем проекте PHP-расширение PCNTL, работники будут обрабатывать POSIX сигнал SIGTERM для окончания обработки текущего сообщения перед выходом.

В некоторых случаях, сигнал SIGTERM отправляется самим Супервизором (например, остановка контейнера Docker в котором Супервизор - точка входа). В таких случаях, вам нужно добавить ключ stopwaitsecs к конфигурации программы (со значением желаемого периода острочки в секундах) для того, чтобы выполнить грациозное выключение:

1
2

[program:x]
stopwaitsecs=20

Работник без состояния¶

PHP создан быть без состояния, разные запросы не имеют общих источников. В HTTP контексте PHP очищает все перед отправкой ответа, поэтому вы можете решить не заботиться о сервисах, которые могут допускать утечку памяти.

С другой стороны, работники обычно работают в долгосрочных CLI-процессах, которые не заканчиваются после обработки сообщения. Поэтому вам нужно быть осторожными с состояниями сервисов, чтобы они не давали утечку информации и/или памяти из одного сообщения в другое.

Однако, некоторые сервисы Symfony, вроде обработчика fingers crossed Monolog, допускают утечку по своей задумке. В таких случаях, используйте опцию транспорта reset_on_message, чтобы автоматически сбросить сервис-контейнер между двумя сообщениями:

Избегание повторных попыток¶

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

Форсирование повторных попыток¶

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

Сохранения и повторные попытки неудачных сообщений¶

Если сообщение терпит неудачу и имеет несколько повторных попыток (max_retries), оно затем будет отбраковано. Чтобы избежать этого, вы можете сконфигурировать failure_transport:

В этом примере, если обработка сообщения терпит неудачу 3 раза (значение по умолчанию max_retries), оно затем будет отправлено транспорту failed. Хотя вы можете использовать messenger:consume failed для потребления его, как обычного транспорта, вам скорее захочется вручную просмотреть сообщения в транспорте ошибок и решить об их повторных попытках:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

# увидеть все сообщения в транспорте ошибок
$ php bin/console messenger:failed:show

# увидеть детали конкретной ошибки
$ 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

Если сообщение опять потерпит неудачу, оно будет отправлено обратно в транспорт ошибок в соответствии с обычными правилами повторных попыток. Как только будет достигнут максимум повторных попыток, сообщение будет сброшено перманентно.

Несколько ошибочных транспортов¶

Иногда недостаточно иметь один глобальный сконфигурированный failed transport, котому что некоторые сообщения важнее, чем другие. В таких случаях, вы можете переопределить транспорт ошибок только для определенных транспортов:

Если нету определенного failure_transport глобально или на уровне транспорта, сообщение будет сброшено после определенного количества попыток.

Неудачные команды имеют необязательную опцию --transport, чтобы указать failure_transport, сконфигурированный на уровне транспорта.

1
2
3
4
5
6
7
8

# увидеть все сообщения в транспорте "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

Транспорт AMQP¶

Транспорт AMQP использует PHP-расширение AMQP для отправки сообщений в очередь вроде RabbitMQ.

1

$ composer require symfony/amqp-messenger

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

1
2
3
4
5

# .env
MESSENGER_TRANSPORT_DSN=amqp://guest:[email protected]:5672/%2f/messages

# или используйте протокол AMQPS
MESSENGER_TRANSPORT_DSN=amqps://guest:[email protected]/%2f/messages

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

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

Транспорт имеет другие опции, включая способы конфигурации обмена, связующие ключи очереди и т.д. Смотрите документацию Connection.

Транспорт имеет ряд опций:

Вы можете также сконфигурирвать настройки специально для AMQP в вашем оообщении, добавив AmqpStamp к вашему Конверту:

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

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

Транспорт Doctrine¶

Транспорт Doctrine может быть использован для хранения сообщений в таблице базы данных.

1

$ composer require symfony/doctrine-messenger

The Doctrine transport DSN may looks like this:

1
2

# .env
MESSENGER_TRANSPORT_DSN=doctrine://default

Формат doctrine://<connection_name>, в случае если у вас есть несколько соединений и вы хотите использовать какое-либо другое, чем “default”. Транспорт будет автоматически создавать таблицу под названием messenger_messages.

Или, для создания таблицы самостоятельно, установите опцию auto_setup как false, и сгенерируйте миграцию.

Транспорт имеет такие опции:

При использовании PostgreSQL, у вас есть доступ к следующим опциям для получения преимуществ функции LISTEN/NOTIFY. Это позволяет более производительный подход, чем поведение голосования транспорта Doctrine по умолчанию, так как PostgreSQL будет напрямую уведомлять работников, когда новое сообщение будет появляться в таблице.

Транспорт Beanstalkd¶

Транспорт Beanstalkd отправляет сообщения прямо с рабочую очередь Beanstalkd. Установите его, выполнив:

1

$ composer require symfony/beanstalkd-messenger

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

1
2
3
4
5

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

# Если порта нет, он по умолчанию будет 11300
MESSENGER_TRANSPORT_DSN=beanstalkd://localhost

Транспорт имеет ряд опций:

Транспорт Redis¶

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

1

$ composer require symfony/redis-messenger

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

1
2
3
4
5
6
7
8

# .env
MESSENGER_TRANSPORT_DSN=redis://localhost:6379/messages
# Полный пример DSN
MESSENGER_TRANSPORT_DSN=redis://[email protected]:6379/messages/symfony/consumer?auto_setup=true&serializer=1&stream_max_entries=0&dbindex=0&delete_after_ack=true
# Пример кластера Redis
MESSENGER_TRANSPORT_DSN=redis://host-01:6379,redis://host-02:6379,redis://host-03:6379,redis://host-04:6379
# Пример Unix-сокета
MESSENGER_TRANSPORT_DSN=redis:///var/run/redis.sock

Некоторые опции могут быть сконфигурированы через DSN или ключ options под транспортом в messenger.yaml:

Транспорт в памяти¶

Транспорт in-memory на самом деле не доставляет сообщения. Вместо этого, он дедржит их в памяти во время запроса, что может быть полезным для тестирования. Например, если у вас есть транспорт async_priority_normal, вы можете переопределить его в окружении test, чтобы использовать этот транспорт:

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

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

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

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

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

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

Транспорт имеет ряд опций:

serialize (булево, по умолчанию: false)

Сериализовать сообщения или нет. Это полезно для тестирования дополнительного слоя, особенно когда вы используете собственный сериализатор сообщений.

Amazon SQS¶

Транспорт Amazon SQS прекрасно подходит для приложения на AWS. Установите его, выполнив:

1

$ composer require symfony/amazon-sqs-messenger

DSN SQS транспорта выглядит так:

1
2
3

# .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

Транспорт имеет ряд опций:

Сериализация сообщений¶

Когда сообщения отправляются (и получаются) в транспорт, они сериализуются с использование нативных функций PHP serialize() и unserialize(). Вы можете изменить это глобально (или для каждого транспорта) на сервис, реализующий SerializerInterface:

messenger.transport.symfony_serializer - это встроенный сервис, который использует компонент Сериализатор и может быть сконфигурирован несколькими способами. Если вы выберете использовать сериализатор Symfony, вы сможете контролировать контекст для каждого случая отдельно через SerializerStamp (см. Конверты и марки).

Manually Configuring Handlers¶

Symfony обычно будет находить и регистрировать вашего обработчика автоматически. Но вы также можете сконфигурировать его вручную - и передать ему дополнительную конфигурацию - тегировав сервис обработчика messenger.message_handler

Возможные опции конфигурации с тегами:

• bus
• from_transport
• handles
• method
• priority

Подписчик и опции обработчика¶

Класс обработчика может обрабатывать множество сообщений или конфигурировать сам себя, реализуя MessageSubscriberInterface:

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

use App\Message\OtherSmsNotification;
use App\Message\SmsNotification;
use Symfony\Component\Messenger\Handler\MessageSubscriberInterface;

class SmsNotificationHandler implements MessageSubscriberInterface
{
    public function __invoke(SmsNotification $message)
    {
        // ...
    }

    public function handleOtherSmsNotification(OtherSmsNotification $message)
    {
        // ...
    }

    public static function getHandledMessages(): iterable
    {
        // обработать это сообщение в __invoke
        yield SmsNotification::class;

        // также обработать это сообщение в handleOtherSmsNotification
        yield OtherSmsNotification::class => [
            'method' => 'handleOtherSmsNotification',
            //'priority' => 0,
            //'bus' => 'messenger.bus.default',
        ];
    }
}

Связывание обработчиков с разными транспортами¶

Каждое сообщение может иметь несколько обработчиков, и когда сообщение потребляется, вызываются все его обработчики. Но вы можете также сконфигурироать обработчика так, чтобы он вызывался только когда сообщение получено из конкретного транспорта. Это позволяет вам имет одно сообщение, где каждый обработчик вызывается разными “работниками”, потребляющими разный транспорт.

Представьте, что у вас есть сообщение UploadedImage с двумя обработчиками:

• ThumbnailUploadedImageHandler: вы хотите, чтобы это обрабатывалось транспортом под названием image_transport
• NotifyAboutNewUploadedImageHandler: вы хотите, чтобы это обрабатывалось транспортом под названием async_priority_normal

Чтобы сделать это, добавьте опцию from_transport к каждому обработчику. Например:

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

use App\Message\UploadedImage;
use Symfony\Component\Messenger\Handler\MessageSubscriberInterface;

class ThumbnailUploadedImageHandler implements MessageSubscriberInterface
{
    public function __invoke(UploadedImage $uploadedImage)
    {
        // создать миниатюры
    }

    public static function getHandledMessages(): iterable
    {
        yield UploadedImage::class => [
            'from_transport' => 'image_transport',
        ];
    }
}

И, похожим образом:

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

class NotifyAboutNewUploadedImageHandler implements MessageSubscriberInterface
{
    // ...

    public static function getHandledMessages(): iterable
    {
        yield UploadedImage::class => [
            'from_transport' => 'async_priority_normal',
        ];
    }
}

Затем, убедитесь, что “маршрутизируете” ваше сообщение к обоим транспортам:

Вот и все! Теперь вы можете потреблять каждый транспорт:

1
2
3
4

# вызовет ThumbnailUploadedImageHandler только при обработке сообщения
$ php bin/console messenger:consume image_transport -vv

$ php bin/console messenger:consume async_priority_normal -vv

Конверты и марки¶

Сообщение может быть любым PHP-объектом. Иногда вам может понадобиться сконфигурировать что-то дополнительное в сообщении - вроде того, как оно должно быть обработано внутри AMQP или добавления задержки перед обработкой сообщения. Вы можете сделать это, добавив марку (“stamp”) к вашему сообщению:

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

public function index(MessageBusInterface $bus)
{
    $bus->dispatch(new SmsNotification('...'), [
        // подождать 5 секунд перед обработкой
        new DelayStamp(5000),
    ]);

    // или ясно создайте Конверт
    $bus->dispatch(new Envelope(new SmsNotification('...'), [
        new DelayStamp(5000),
    ]));

    // ...
}

Внутренне, каждое сообщение оборачивается в конверт (Envelope), содержащий сообщение и марки. Вы можете создать его вручную или позволить автобусу сообщений сделать это. Существует множество различных марок для разных целей и они используются внутренне для отслеживания информации о сообщении - вроде того, какой автобус обрабатывает его или имеет ли оно повторные попытки после неудачи.

Промежуточное ПО¶

То, что происходит после запуска сообщения в автобус сообщений, зависит от его набора промежуточного ПО и его порядка. По умолчанию, промежуточное ПО, сконфигурированное для каждого автобуса, выглядит так:

1. add_bus_name_stamp_middleware - добавляет марку для записи того, в каком атобусе было запущено это собщение;
2. dispatch_after_current_bus- см. Transactional Messages: Handle New Messages After Handling is Done;
3. failed_message_processing_middleware - обрабатывает сообщения, которые имеют повторные попытки через транспорт ошибок, чтобы они правильно функционировали, как будто бы они были получены из изначального транспорта;
4. Ваша собственная коллекция middleware_;
5. send_message - если машрутизация сконфигурирована для транспорта, отправляет сообщения этому транспорту и останавливает цепь промежуточного ПО;
6. handle_message - вызывает обработчика(ов) сообщений для заданног сообщения.

Промежуточное ПО выполняется после запуска сообщения, и также еще раз, когда сообщение получено через работника (для сообщений, которые были отправлены транспорту для асинхронной обработки). Помните это, если вы создаете собственное промежуточное ПО.

Вы можете добавить собственное промежуточное ПО в список, или полностью отключить промежуточное ПО по умолчанию и добавить только ваше собственное:

Промежуточное ПО для Doctrine¶

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

Другое промежуточное ПО¶

Добавьте промежуточное ПО router_context, если вам нужно генерировать абсолютные URL в потребителе (например, отображать шаблон со ссылками). Это промежуточное ПО хранит контекст изначального запроса (т.е. хост, HTTP-порт и т.д.), что необходимо при создании абсолютных URL.

События Мессенджера¶

В дополнение к промежуточному ПО, Мессенджер также запускает несколько событий. Вы можете создать слушателя событий, чтобы подключаться к разным частям процесса. Для каждого, класс события будет названием события:

• WorkerStartedEvent
• WorkerMessageReceivedEvent
• SendMessageToTransportsEvent
• WorkerMessageFailedEvent
• WorkerMessageHandledEvent
• WorkerRunningEvent
• WorkerStoppedEvent