Новое в Symfony 5.4: Улучшения компонента Messenger

Настраиваемый обработчик с PHP атрибутами

Атрибуты PHP - отличный способ добавления метаданных в PHP код. В Symfony мы добавили возможность использования атрибутов PHP для конфигурирования множества вещей. Вот почему в Symfony 5.4 мы позволили настраивать обработчик сообщений при помощи атрибутов.

Вместо реализации MessageHandlerInterface теперь вы можете добавить атрибут AsMessageHandler к любому PHP классу и использовать его в качестве обработчика сообщений:

// 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)
    {
        // ...
    }
}

Метаданные воркера

Сейчас нет простого способа получить метаданные воркера, такие, как имя его транспорта. В Symfony 5.4 мы улучшаем это с введением нового класса, который доступен через $worker->getWorkerMetadata(). Например, внутри метода некоторого слушателя/подписчика, который обрабатывает события Symfony Messenger, вы можете использовать что-то вроде этого:

public function resetServices(WorkerRunningEvent $event): void
{
    $actualTransportName = $event->getWorker()->getWorkerMetadata()->getTransportName();
    if (!$event->isWorkerIdle() || !in_array($actualTransportName, $this->receiversName, true)) {
        return;
    }

    $this->servicesResetter->reset();
}

Сброс сервисов контейнера между сообщениями

Сервисы контейнера не сбрасываются автоматически при обработке сообщений. Это может быть проблемой, например, с обработчиком "со скрещёнными пальцами" Monolog. Поскольку сервисы не сбрасываются, если первое сообщение вызывает ошибку, следующие сообщения будут логироваться и в конечном итоге переполнят буфер.

В Symfony 5.4 мы улучшили ситуацию с помощью опции автоматического сброса сервисов после обработки сообщения. Чтобы использовать эту возможность, установите в конфигурации вашего мессенджера опции new reset_on_message option значение true:

# config/packages/messenger.yaml
framework:
    messenger:
        reset_on_message: true
        transports:
            async:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
            failed: 'doctrine://default?queue_name=failed'
            sync: 'sync://'

Обработка сообщений в пакетном режиме

Иногда при использовании компонента Messenger вы могли бы обрабатывать несколько сообщений одновременно, а не по одному друг за другом. В Symfony 5.4 мы представили новый интерфейс BatchHandlerInterface, который позволяет вашим обработчикам обрабатывать сообщения партиями.

Обработчик реализующий этот интерфейс должен ожидать новый опциональный аргумент $ack, который будет предоставлен при вызове __invoke(). Если вы не укажите аргумент $ack, сообщение будет обрабатываться синхронно, как обычно. Если вы предоставляете $ack, ожидается, что __invoke() буферизирует сообщение и его функция $ack вернёт количество ожидающих сообщений в пакете. Вот как может выглядеть пакетный обработчик:

class MyBatchHandler implements BatchHandlerInterface
{
    use BatchHandlerTrait;

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

    private function process(array $jobs): void
    {
        foreach ($jobs as [$message, $ack]) {
            try {
                // [...] compute $result from $message
                $ack->ack($result);
            } catch (\Throwable $e) {
                $ack->nack($e);
            }
        }
    }
}

Размер пакета сообщений контролируется BatchHandlerTrait::shouldFlush() (по умолчанию 10)