Transports

Where a message waits between dispatch and handling: a DSN, a routing entry, and a worker reading the queue.

Syntaxphp bin/console messenger:consume async --time-limit=3600 --memory-limit=128M

A transport is the place a message sits between being dispatched and being handled. Each transport is named in config/packages/messenger.yaml and configured with a DSN, a connection string. Routing maps a message class to a transport name.

The three you will meet

doctrine://default keeps messages in a database table, messenger_messages by default, using the connection you already have. It needs no extra server and is usually the right first choice. redis://localhost:6379/messages uses a Redis stream and holds up better under volume. amqp://guest:guest@localhost:5672/%2f/messages talks to RabbitMQ; the %2f is an encoded /, the default virtual host.

Consuming

A worker is a long-running process that reads one transport and calls handlers: php bin/console messenger:consume async. It loads your code once at boot and never sees an edit, so give it limits such as --time-limit=3600 and --memory-limit=128M and let something restart it.

Shorthand and longhand

async: '%env(MESSENGER_TRANSPORT_DSN)%' is shorthand for the mapping form async: {dsn: '%env(MESSENGER_TRANSPORT_DSN)%'}. They configure the same transport. Switch to the long form the moment you need options, retry_strategy or failure_transport.

What trips people up

The Doctrine transport creates its table on the first dispatch. Convenient locally, wrong in production, where the application user may have no rights to alter the schema. Set auto_setup=false in the DSN and create the table in a migration instead.

Example

Example · php
// src/Service/ExportDispatcher.php
declare(strict_types=1);

namespace App\Service;

use App\Message\ExportOrders;
use Symfony\Component\Messenger\MessageBusInterface;
use Symfony\Component\Messenger\Stamp\TransportNamesStamp;

final class ExportDispatcher
{
    public function __construct(private readonly MessageBusInterface $bus)
    {
    }

    /**
     * Small exports go wherever routing sends them. Large ones go to a
     * separate transport with its own workers, so one hour-long export
     * cannot hold up everything behind it.
     */
    public function dispatch(int $customerId, int $rowCount): void
    {
        $message = new ExportOrders($customerId);

        if ($rowCount < 1_000) {
            // No stamp: the transport comes from the routing configuration.
            $this->bus->dispatch($message);

            return;
        }

        // TransportNamesStamp takes an array of transport names and overrides
        // routing for this one dispatch.
        $this->bus->dispatch($message, [new TransportNamesStamp(['async_bulk'])]);
    }
}

When to use it

  • Start on the Doctrine transport so a new project needs no queue server.
  • Give slow jobs their own transport and their own workers.
  • Move from the database to Redis by changing one environment variable.
  • Run two workers on different queues on the same host.

More examples

Two transports and their routing

Separate transports exist so you can give them separate workers. Dispatching is unchanged; only the queue a message lands in differs. The queue_name option here is the Doctrine transport's, which keeps both queues in one messenger_messages table.

Example · yaml
# config/packages/messenger.yaml
framework:
    messenger:
        transports:
            # Shorthand: a DSN string.
            async: '%env(MESSENGER_TRANSPORT_DSN)%'

            # Longhand: the same thing as a mapping, which is what you need
            # as soon as there are options to set.
            async_bulk:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                options:
                    queue_name: bulk

        routing:
            App\Message\SendWelcomeEmail: async
            App\Message\ExportOrders: async_bulk

The DSNs, in .env

One environment variable is usually enough to move an application from the database transport in development to Redis or RabbitMQ in production. Keep the value in the environment, not in messenger.yaml, so each deployment can differ.

Example · bash
# .env
# Database-backed: no extra service to run.
MESSENGER_TRANSPORT_DSN=doctrine://default

# Same, with table creation turned off for production. Create the table in a
# migration instead of letting the first dispatch do it.
# MESSENGER_TRANSPORT_DSN=doctrine://default?auto_setup=false

# Redis stream: stream name, consumer group, then a per-worker consumer name.
# MESSENGER_TRANSPORT_DSN=redis://localhost:6379/messages/symfony/consumer-1

# RabbitMQ: %2f is the encoded default virtual host "/".
# MESSENGER_TRANSPORT_DSN=amqp://guest:guest@localhost:5672/%2f/messages

Pinning a handler to one transport

Useful when the same message is routed to more than one transport and each should do something different, or when you want a guarantee that heavy work only ever runs on the machines dedicated to it. The attribute also accepts bus, handles, method and priority.

Example · php
// src/MessageHandler/ExportOrdersHandler.php
declare(strict_types=1);

namespace App\MessageHandler;

use App\Export\OrderExporterInterface;
use App\Message\ExportOrders;
use Symfony\Component\Messenger\Attribute\AsMessageHandler;

// fromTransport limits this handler to messages consumed from async_bulk.
// A worker reading another transport will not call it, even for this message.
#[AsMessageHandler(fromTransport: 'async_bulk')]
final class ExportOrdersHandler
{
    public function __construct(private readonly OrderExporterInterface $exporter)
    {
    }

    public function __invoke(ExportOrders $message): void
    {
        $this->exporter->export($message->customerId);
    }
}

Related

symfony messenger introsymfony messenger reliabilitysymfony messenger patternssymfony env varssymfony k8s workers

Discussion

  • Be the first to comment on this lesson.