Транспорт failed

Транспорт failed - это специальный транспорт, который принимает сообщения, не обработанные успешно после исчерпания всех повторных попыток. Он настраивается в секции framework.messenger.transports и позволяет сохранять «проваленные» сообщения для последующего анализа или повторной обработки. По умолчанию используется транспорт doctrine, но можно указать любой другой DSN.

Синтаксис

framework:
	messenger:
		failure_transport: failed
		transports:
			failed: 'doctrine://default?queue_name=failed'

Пример

Давайте настроим транспорт failed в файле конфигурации:

framework:
	messenger:
		failure_transport: failed
		transports:
			async: '%env(MESSENGER_TRANSPORT_DSN)%'
			failed: 'doctrine://default?queue_name=failed'
		routing:
			App\Message\ArticleMessage: async

Теперь при неудачной обработке сообщение попадёт в очередь failed.

Пример

Давайте создадим обработчик, который намеренно выбрасывает исключение, чтобы сообщение попало в транспорт failed:

<?php
	namespace App\MessageHandler;
	
	use App\Message\ArticleMessage;
	use Symfony\Component\Messenger\Attribute\AsMessageHandler;
	
	#[AsMessageHandler]
	class ArticleMessageHandler
	{
		public function __invoke(ArticleMessage $message): void
		{
			throw new \RuntimeException('article not processed');
		}
	}
?>

После запуска воркера сообщение будет помечено как неудачное и перемещено в очередь failed:

php bin/console messenger:consume async

Пример

Давайте посмотрим список неудачных сообщений в очереди failed:

php bin/console messenger:failed:show

Результат выполнения команды:

1 message(s) in the failure transport.
------------------------ ------------------------ 
 Id					   Class
------------------------ ------------------------
 1						App\Message\ArticleMessage

Пример

Давайте повторно обработаем все неудачные сообщения из очереди failed:

php bin/console messenger:failed:retry

Результат выполнения команды:

Retrying message 1...
"article not processed"

Смотрите также

транспорт async,
который обрабатывает сообщения асинхронно
транспорт sync,
который обрабатывает сообщения синхронно
класс Worker,
который запускает обработку сообщений из очереди
класс HandlerFailedException,
которое выбрасывается при ошибке обработчика