Using Symfony Messenger to Manage Message Queues in Symfony

Symfony, the famous PHP framework, offers a wide range of powerful tools and components for developers of all levels. One of these tools is Symfony Messenger, a bundle that greatly simplifies message handling within Symfony applications.

What is Symfony Messenger?

Symfony Messenger is a bundle that provides everything you need to consume messages from a message queue. This approach is extremely useful for improving the performance of our applications by allowing us to separate heavy tasks and handle them through a separate worker.

Advantages of Message Queuing

Message queuing allows you to:

• Improve application performance.

• Handle heavy tasks asynchronously.

• Maintain a fast response time for users.

If you are not familiar with message queuing, I recommend delving into the topic to fully understand its benefits.

Use Case

To better understand the usefulness of Symfony Messenger, let's consider a common use case: managing orders and shipments in an application.

Let's imagine having to perform several tasks every time an order is updated, such as:

• Updating the order status on the website.

• Sending a confirmation email to the user.

• Perhaps, sending a notification via SMS.

• Executing other internal scripts.

All these tasks require calls to different services, which can take a lot of time, and our goal is to have the shortest response time possible.

By using Symfony Messenger, we could delegate these tasks to a separate worker, significantly improving the overall performance of the application.

Introduction and Installation

To start using Symfony Messenger, we need to first install the package via Composer:

composer require symfony/messenger

This package provides everything needed to create and manage messages and message handlers within our application.

Messages and Handlers

The core of Symfony Messenger revolves around messages and message handlers. A message represents the data to be processed, while a message handler contains the logic to process that message.

For example, suppose we need to send a registration confirmation email to a user who has just registered on our site. We will create a message called SendRegistrationEmailMessage which will contain the user ID.

<?php

namespace App\Message;

class SendRegistrationEmailMessage
{
    public function __construct(
        private readonly int $userId
    ) {}

    public function getUserId(): int
    {
        return $this->userId;
    }
}

Instead of executing everything at the exact moment the registration request is sent, we will queue a message containing the user ID. This way, our controller no longer needs to handle an exception, and we could send a "OK" (200) response much faster.

Next, we will create a message handler called SendRegistrationEmailMessageHandler that will retrieve the user from the ID and send the confirmation email.

<?php

namespace App\MessageHandler;

use App\Entity\User;
use Symfony\Component\Messenger\Attribute\AsMessageHandler;

#[AsMessageHandler]
class SendRegistrationEmailMessageHandler
{
    public function __construct(
        private readonly EntityManagerInterface $entityManager
    ) {}

    public function __invoke(SendRegistrationEmailMessage $message): void
    {
        $user = $this->entityManager->getRepository(User::class)->find($message->getUserId());

        if (!$user instanceOf User) {
            // Handle entity not found case
        }

        // Send the email
    }
}

Symfony Messenger simplifies this process, allowing us to focus on business logic without worrying about implementation details.

Let's see how.

Symfony Messenger comes with a PHP attribute #[AsMessageHandler], so that Symfony considers this service as a message handler to route the correct messages to.

The type of the message to consume is declared in the __invoke method.
This means that for each message, we should create a message handler, with the correct type in the __invoke method, and the routing will be handled automatically by Symfony.

Message Transport and Routing

Symfony Messenger supports various types of transports:

• AMQP services (like RabbitMQ).

• Doctrine (storing messages in an SQL table).

• Cache services (like Redis).

First of all, if we intend to use an AMQP protocol, we need to install the package:

> composer require symfony/amqp-messenger

If we intend to use Doctrine, we need to install the following:

> composer require symfony/doctrine-messenger

Finally, if we want to use Redis as our transport, this is the package we need to install:

> composere require symfony/redis-messenger

If you want to learn more about other transports, you can take a look at the relevant documentation.

We can easily configure the desired transport in the messenger.yaml configuration file, which we will find in config/packages along with all the other configuration files of installed packages.

The transport key contains all the configuration regarding message handling (consumption).

The failure-transport key contains the name of the transport to be used in case of issues (exception thrown during handling).
If something goes wrong here, the stack trace and exception details will be saved in the message and can be retrieved later to be properly handled.

We can also configure the consumer to perform x retries before sending an error message to the "failed" queue and manage different priorities for each queue.

Finally, the routing part explains how to route a specific instance of the message to the transport.

Here is an example of how our configuration file for the above case might look:

framework:
    messenger:
        transports:
            registration_email:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                failure_transport: registration_email_failed
                retry_strategy:
                    max_retries: 3
                    delay: 1000
                    multiplier: 2
                    max_delay: 0
                options:
                    exchange:
                        name: registration_email
                    queues:
                        registration_email: ~

            registration_email_failed:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                options:
                    exchange:
                        name: registration_email_failed
                    queues:
                        registration_email_failed: ~

        routing:
            'App\Message\SendRegistrationEmailMessage': registration_email

Dispatching the Message

Dispatching a message is as simple as calling the MessageBus service provided by Symfony Messenger and passing the message to dispatch. The bus will take care of the rest, routing the message to the appropriate handler for processing.

Here's an example:

<?php

namespace App\Service;

use App\Message\SendEMailRegistrationEmailMessage;

class MyExampleService
{
    public function __construct(
        private readonly MessageBusInterface $messageBus
    ) {}

    public function doSomething(): void
    {
        // Any logic...

        $message = new SendEMailRegistrationEmailMessage($userId);

        # This will dispatch the message to the correct transport
        $this->messageBus->dispatch($message);
    }
}

Architecture

The architecture of Symfony Messenger is well-structured and involves the use of a publisher (controller, service, command, etc.) that sends a message to the bus. If the bus is synchronous, the message is consumed directly by a handler. If the bus is asynchronous, the message is sent via a transport to a queuing system, where it is processed by a separate worker.

If the transport is asynchronous, the message must be serialized in order to be interpreted and consumed correctly by the worker. Symfony natively supports two serialization modes:

• PHP's native serialization.

• Serialization via the Serializer component.

By default, PHP's native serialization is used, which represents the class itself of the message.

If another application consumes the same message queue, it may not have this class, so it will be impossible to deserialize the message. For this reason, we will use a more traditional exchange format. Usually JSON, but we could use XML, Protobuf, or any other interoperable serialization language.

Conclusion

Symfony Messenger is a powerful tool for message handling within Symfony applications. With its support for various transport types and its simple architecture, it allows us to improve application performance and provide a better user experience.

I hope this article has provided you with a comprehensive overview of Symfony Messenger and inspired you to use it in your future Symfony applications! If you want to receive updates on future articles, feel free to follow my account on Medium!

Good work! 👨‍💻