◀ Back to blog
PHP / Symfony

Asynchronous processing with Symfony Messenger

Published on 19 Jul 2024· 7 min read
#Symfony#Messenger#Async

Symfony Messenger: asynchronous processing made simple

Symfony's Messenger component lets you decouple long-running work from your HTTP requests for a better user experience.

Sending an email, generating a PDF, calling a slow third-party API or resizing an image: none of these tasks needs to block the response sent to the user. With Messenger, the controller saves what matters, publishes a message to a queue and responds immediately. A separate process, the worker, then handles the message in the background.

The building blocks of Messenger

  • The message: a plain PHP object carrying the data needed for the work. It contains no logic.
  • The handler: the class that knows how to process one type of message.
  • The bus: the entry point, MessageBusInterface, to which you hand messages with dispatch().
  • The transport: the queue where messages wait (RabbitMQ, Redis, a Doctrine table, Amazon SQS…).
  • The worker: the messenger:consume command, which reads the transport and calls the handlers.

Without any routing configuration, a message is handled synchronously, inside the request. Routing it to a transport is what makes processing asynchronous, without touching the message or handler code.

Message/Handler architecture

// Message
class SendNotificationMessage
{
    public function __construct(
        public readonly int $userId,
        public readonly string $content,
    ) {}
}

// Handler
#[AsMessageHandler]
class SendNotificationHandler
{
    public function __construct(
        private NotificationService $notificationService,
    ) {}

    public function __invoke(SendNotificationMessage $message): void
    {
        $this->notificationService->send(
            $message->userId,
            $message->content,
        );
    }
}

The message is immutable (readonly) and only holds scalars: it will be serialized to be stored in the queue, then deserialized by the worker, sometimes several minutes later. Pass an identifier rather than a Doctrine entity: the handler will reload a fresh copy from the database. The #[AsMessageHandler] attribute is enough for Symfony to link the handler to the message, based on the type of the __invoke() argument.

Transport configuration

# config/packages/messenger.yaml
framework:
    messenger:
        failure_transport: failed

        transports:
            async:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                retry_strategy:
                    max_retries: 3
                    delay: 1000
                    multiplier: 2
            failed:
                dsn: 'doctrine://default?queue_name=failed'

        routing:
            App\Message\SendNotificationMessage: async
            App\Message\ProcessOrderMessage: async

The retry strategy is expressed in milliseconds: when an exception is thrown, the message is retried after 1 second, then 2, then 4. After the third failure, it goes to the transport named by failure_transport, here a Doctrine table. Without that key, a message that has used up its retries is simply lost.

The DSN of the main transport depends on your infrastructure:

# .env
# RabbitMQ (requires the amqp PHP extension)
MESSENGER_TRANSPORT_DSN=amqp://guest:guest@localhost:5672/%2f/messages
# Redis
# MESSENGER_TRANSPORT_DSN=redis://localhost:6379/messages
# Database, no extra infrastructure
# MESSENGER_TRANSPORT_DSN=doctrine://default?auto_setup=0

The Doctrine transport is a good starting point: it needs nothing beyond the existing database. RabbitMQ or Redis become worthwhile when message volume grows or when several applications share the queues.

Dispatching messages

class OrderController extends AbstractController
{
    public function __construct(
        private OrderService $orderService,
    ) {}

    #[Route('/order', methods: ['POST'])]
    public function create(
        MessageBusInterface $bus,
        Request $request,
    ): JsonResponse {
        // Fast synchronous work
        $order = $this->orderService->create($request);

        // Asynchronous work
        $bus->dispatch(new SendNotificationMessage(
            $order->getUserId(),
            "Order #{$order->getId()} confirmed"
        ));

        return $this->json($order, 201);
    }
}

Creating the order stays synchronous: the user needs to know right away whether it succeeded. Only the notification goes to the queue. Dispatch the message after saving to the database: if the worker handles it before the transaction ends, it will not find the order.

Running and supervising workers

In development, a terminal is enough:

php bin/console messenger:consume async -vv

# In production: restart the worker regularly
php bin/console messenger:consume async --time-limit=3600 --memory-limit=256M

A PHP worker is a long-running process: it does not reload the code after a deployment and its memory can grow. The --time-limit and --memory-limit options stop it cleanly, after the current message, and the process manager restarts it straight away. With systemd, a template unit lets you run several workers:

# /etc/systemd/system/messenger-worker@.service
[Unit]
Description=Symfony Messenger worker %i
After=network.target

[Service]
User=app
WorkingDirectory=/var/www/app
ExecStart=/usr/bin/php bin/console messenger:consume async --time-limit=3600 --memory-limit=256M
Restart=always
RestartSec=5

[Install]
WantedBy=multi-user.target
sudo systemctl daemon-reload
sudo systemctl enable --now messenger-worker@1 messenger-worker@2

On every deployment, run php bin/console messenger:stop-workers: workers finish their current message, stop, and systemd restarts them with the new code.

Handling failed messages

php bin/console messenger:failed:show
php bin/console messenger:failed:retry
php bin/console messenger:failed:remove 42

Once a bug is fixed, messenger:failed:retry replays the stuck messages. If an error is permanent (deleted user, invalid data), throw an UnrecoverableMessageHandlingException in the handler: the message will not be retried for nothing.

Common pitfalls

  • Non-idempotent handlers: Messenger guarantees "at least once" delivery. A message can be handled twice after a crash; the handler must cope without sending two payments or two emails.
  • Lost database connection: a worker that stays idle for a long time can have its MySQL connection closed by the server. The doctrine_ping_connection middleware checks it before each message.
  • Heavy messages: put no entities, files or services in them. Identifiers and scalars are enough.
  • Tests: in the test environment, route to the in-memory:// transport to assert that a message was dispatched, or to sync:// to run it immediately.

Supervising workers

  • Use systemd to manage workers in production
  • Set --time-limit to avoid memory leaks
  • Monitor the failed queue for messages in error
  • Use the logging middleware for debugging
  • Stop workers cleanly on every deployment with messenger:stop-workers
  • Track queue sizes with messenger:stats to spot a stuck worker

When not to go asynchronous

If the user needs the result to continue their journey (a payment to validate, stock to reserve), the processing must stay synchronous or come with a tracking mechanism, such as a status polled by the front end. Asynchronous processing also adds infrastructure to monitor: for a task that takes a few milliseconds, it brings nothing. Keep it for tasks that are slow, fragile or not essential to the response.

In short: small immutable messages, idempotent handlers, a configured failure transport, workers restarted regularly and stopped cleanly on every deployment. With these few rules, Messenger becomes a reliable foundation for absorbing load spikes and keeping response times short.