RabbitMQ is an optional dependency used for async message queuing. The framework provides two classes — RabbitMQ for dispatching messages and RabbitMQConsumer for consuming them — both driven by QueueEnum which defines the available queues and ties them to the Symfony Messenger transport configuration.
RabbitMQ support in the framework is split between two layers with different dependencies — install only what you need.
RabbitMQ / RabbitMQConsumer classesphp-amqplib/php-amqplib Composer package — pure-PHP AMQP 0-9-1 client (included in the template)PHP_SF\System\Database\RabbitMQ (dispatch) and PHP_SF\System\Database\RabbitMQConsumer (consume)messenger.yaml but build their own AMQP connection — they do not go through Symfony Messenger's transportsymfony/amqp-messenger Composer packageext-amqp PHP extension — wraps the librabbitmq C client; install via your package manager (apt install php-amqp, pecl install amqp) and enable in php.inirouting: { 'Symfony\Component\Mailer\Messenger\SendEmailMessage': emails })Both layers can coexist in the same project. php-amqplib and ext-amqp are not interchangeable — they back different transport stacks.
MESSENGER_TRANSPORT_DSN set in .envframework.messenger.transports in config/packages/messenger.yaml — the framework reads this file at runtime to validate that each QueueEnum case value matches a declared transport nameIf you're not using async queues, none of this is needed and RabbitMQ can be left out entirely.
MESSENGER_TRANSPORT_DSN=amqp://admin:admin@127.0.0.1:7004/%2f
The DSN format is amqp://user:password@host:port/vhost. The default vhost in RabbitMQ is /, which URL-encodes to %2f.
Queues are defined as Symfony Messenger transports in config/packages/messenger.yaml:
framework:
messenger:
failure_transport: async
transports:
async: 'in-memory://'
default_queue: '%env(MESSENGER_TRANSPORT_DSN)%/default_queue'
routing:
'stdClass': default_queue
The template ships with two transports: async (in-memory, for tests) and default_queue (AMQP-backed, the actual production queue). The framework expects every QueueEnum case value to match one of these transport names — otherwise RabbitMQ::getInstance() throws InvalidRabbitMQConfigurationException on first use.
QueueEnum is the central place to define and access queues. It's a backed enum where each case value is the queue name as declared in messenger.yaml:
// App/Enums/Amqp/QueueEnum.php
enum QueueEnum: string
{
case DEFAULT = 'default_queue';
// todo: add more queues, if needed
public function getMessageBus(): RabbitMQ
{
return RabbitMQ::getInstance( $this );
}
}
Each case value must match a transport declared in messenger.yaml — RabbitMQ::parseConfig() indexes transports by name on construction and setQueue() rejects any case whose value is not in that index, throwing InvalidRabbitMQConfigurationException("Queue {value} not found in config/packages/messenger.yaml"). The framework's RabbitMQ::getInstance() defaults to QueueEnum::DEFAULT, so that transport must exist for the default constructor call to work.
Adding a new queue means two steps — adding a case to QueueEnum and adding the matching transport to messenger.yaml. The host project owns App\Enums\Amqp\QueueEnum; the framework's RabbitMQ / RabbitMQConsumer import it directly, so the enum lives in the host app namespace (not the framework's).
Use QueueEnum to get a RabbitMQ instance and dispatch:
// Dispatch a JSON string to the default queue
QueueEnum::DEFAULT
->getMessageBus()
->dispatch( json_encode( [
'type' => 'item_action',
'user_id' => $userId,
'action' => 'update',
'data' => $actionData,
] ) );
dispatch() accepts a string. Serialize your payload before passing it in — JSON is the recommended format.
RabbitMQ is a singleton per queue — RabbitMQ::getInstance( QueueEnum::DEFAULT ) always returns the same connection for that queue within a request. The AMQP connection and channel are opened once on first call and closed in __destruct().
Consumption is a blocking operation intended for long-running console commands, not HTTP requests. Create a Symfony console command and call RabbitMQConsumer::consume() with the queue case and a callback:
// App/Command/ProcessMessagesCommand.php
#[AsCommand(
name: 'app:queue:process',
description: 'Process incoming messages',
)]
final class ProcessMessagesCommand extends Command
{
protected function execute( InputInterface $input, OutputInterface $output ): int
{
$io = new SymfonyStyle( $input, $output );
$io->info( 'Listening for messages...' );
( new RabbitMQConsumer() )->consume( QueueEnum::DEFAULT, function ( array $data ) use ( $io ): void {
$io->text( sprintf(
'Processing action "%s" for user %d',
$data['action'],
$data['user_id']
) );
// handle the message
} );
return Command::SUCCESS;
}
}
The callback receives the decoded JSON payload as an array. RabbitMQConsumer calls json_decode() on the raw message body before passing it to your callback, so you always receive an array rather than a raw string.
The consumer blocks indefinitely in while ( $this->channel->is_consuming() ) until the process is killed or the connection drops. Run it under a process supervisor (systemd, Supervisor, etc.) in production.
Step 1 — Add the transport to messenger.yaml:
transports:
async: 'in-memory://'
default_queue: '%env(MESSENGER_TRANSPORT_DSN)%/default_queue'
app_events: '%env(MESSENGER_TRANSPORT_DSN)%/app_events'
Step 2 — Add the case to QueueEnum:
enum QueueEnum: string
{
case Default = 'default_queue';
case AppEvents = 'app_events';
}
Step 3 — Use it:
QueueEnum::AppEvents
->getMessageBus()
->dispatch( json_encode( $eventPayload ) );
RabbitMQ is a singleton per queue. On first instantiation for a given QueueEnum case it:
config/packages/messenger.yaml to validate the queue existsMESSENGER_TRANSPORT_DSN to extract host, port, user, passwordAMQPStreamConnectiondurable: trueThe connection and channel are stored as instance properties and reused for all dispatch() calls on the same queue within a request. Both are closed cleanly in __destruct().
// Internal connection parsing — handled automatically
$dsn = parse_url( env( 'MESSENGER_TRANSPORT_DSN' ) );
// host, port, user, pass extracted from DSN
Unlike RabbitMQ, RabbitMQConsumer is not a singleton — instantiate it fresh each time you need to consume. It opens its own connection and channel, declares the same queue, and sets up a basic consumer with basic_consume().
The internal callback wraps your callable and handles JSON decoding:
$internalCallback = function ( AMQPMessage $msg ) use ( $callback ): void {
$data = json_decode( $msg->getBody(), true );
$callback( $data );
};
auto_ack is set to true — messages are acknowledged automatically on delivery. If your callback throws, the message will not be requeued. Add your own try/catch inside the callback if you need error handling or dead-letter routing.
Process supervision — consumer commands must be kept alive by a process manager. If the command exits (connection drop, OOM kill, deploy restart), messages will queue up in RabbitMQ until the consumer restarts. Use systemd or Supervisor with autorestart=true.
Message acknowledgement — the current implementation uses auto-ack. For critical workloads where message loss is unacceptable, you'll need to modify RabbitMQConsumer to use manual acknowledgement (auto_ack: false) and call $msg->ack() after successful processing.
Failure transport — messenger.yaml sets failure_transport: async which routes failed messages back to the async (in-memory) transport. For production you likely want a dedicated dead-letter queue instead.
Multiple consumers — you can run multiple instances of the same consumer command in parallel. RabbitMQ will round-robin messages between them automatically.
Queue name mismatch — the QueueEnum case value must exactly match the transport name in messenger.yaml. A mismatch throws InvalidRabbitMQConfigurationException with a clear message pointing to the yaml file.
Dispatching in a consumer callback — dispatching a message from inside a consumer callback on the same queue will work but can cause processing loops if the dispatch is unconditional. Always guard dispatch calls inside callbacks with a condition.
Running consumers in HTTP requests — consume() blocks indefinitely. Never call it in a controller or middleware. It belongs exclusively in console commands.
Missing MESSENGER_TRANSPORT_DSN — if the env var is not set, parse_url() will receive null and the connection will fail with a confusing error. The env var must be set even if you define it with a default in the class — the class default (amqp://guest:guest@localhost:7004) is a framework fallback and may not match your Docker setup (the template project uses admin:admin).