Magento 2 RabbitMQ Consumers That Don't Die Silently

Message queues in Magento are easy to set up and easy to break without noticing. A consumer stops, nothing throws an error on the storefront, and days later someone asks why orders were not exported or why bulk actions in the admin never finished. The queue just kept growing in the background.

This article covers how the queue configuration fits together in Magento 2.4, how consumers are started, and the settings and process management that keep them running. Everything here applies to Magento Open Source and Adobe Commerce with RabbitMQ as the broker.

Connecting Magento to RabbitMQ

The AMQP connection lives in app/etc/env.php. You can write it with the setup command instead of editing the file by hand:

bin/magento setup:config:set \
  --amqp-host=rabbitmq \
  --amqp-port=5672 \
  --amqp-user=magento \
  --amqp-password='change-me' \
  --amqp-virtualhost=/

Without this block, Magento uses its MySQL queue where it can. Check which backend you are actually using before debugging anything else, because the tools and the symptoms are different.

The four XML files

A queue in Magento is described by four files in the module etc folder. Each one answers a different question.

communication.xml: what is the message

Declares the topic name and the type of data it carries. The request type can be a scalar such as string or an interface with getters and setters, which Magento serializes for you.

<?xml version="1.0"?>
<config xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:noNamespaceSchemaLocation="urn:magento:framework:Communication/etc/communication.xsd">
    <topic name="vendor.order.export" request="string"/>
</config>

queue_publisher.xml: where does it go

Tells Magento which connection and exchange a topic is published to.

<?xml version="1.0"?>
<config xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:noNamespaceSchemaLocation="urn:magento:framework-message-queue:etc/publisher.xsd">
    <publisher topic="vendor.order.export">
        <connection name="amqp" exchange="magento"/>
    </publisher>
</config>

queue_topology.xml: how is it routed

Binds the topic on the exchange to a queue. Magento creates the exchange, the queue and the binding in RabbitMQ when you run setup:upgrade.

<?xml version="1.0"?>
<config xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:noNamespaceSchemaLocation="urn:magento:framework-message-queue:etc/topology.xsd">
    <exchange name="magento" type="topic" connection="amqp">
        <binding id="vendorOrderExportBinding"
                 topic="vendor.order.export"
                 destinationType="queue"
                 destination="vendor.order.export"/>
    </exchange>
</config>

queue_consumer.xml: who reads it

Names the consumer, the queue it reads and the handler method that processes each message.

<?xml version="1.0"?>
<config xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:noNamespaceSchemaLocation="urn:magento:framework-message-queue:etc/consumer.xsd">
    <consumer name="vendor.order.export"
              queue="vendor.order.export"
              connection="amqp"
              maxMessages="1000"
              consumerInstance="Magento\Framework\MessageQueue\Consumer"
              handler="Vendor\OrderExport\Model\Queue\ExportHandler::process"/>
</config>

Run bin/magento queue:consumers:list after setup:upgrade to confirm the consumer is registered.

Publishing and handling messages

Publishing is a single call on Magento\Framework\MessageQueue\PublisherInterface. Publish ids, not whole objects. The entity may change between publish and consume, and small messages are cheap to retry.

<?php
declare(strict_types=1);

namespace Vendor\OrderExport\Observer;

use Magento\Framework\Event\Observer;
use Magento\Framework\Event\ObserverInterface;
use Magento\Framework\MessageQueue\PublisherInterface;

class QueueOrderExport implements ObserverInterface
{
    public function __construct(
        private readonly PublisherInterface $publisher
    ) {
    }

    public function execute(Observer $observer): void
    {
        $order = $observer->getEvent()->getOrder();
        $this->publisher->publish('vendor.order.export', (string)$order->getId());
    }
}

The handler receives the decoded message. Make it idempotent, because a message can be delivered more than once, for example when the process dies after the work is done but before the acknowledgement reaches the broker.

<?php
declare(strict_types=1);

namespace Vendor\OrderExport\Model\Queue;

use Psr\Log\LoggerInterface;
use Vendor\OrderExport\Model\Exporter;

class ExportHandler
{
    public function __construct(
        private readonly Exporter $exporter,
        private readonly LoggerInterface $logger
    ) {
    }

    public function process(string $orderId): void
    {
        if ($this->exporter->isExported((int)$orderId)) {
            return;
        }

        try {
            $this->exporter->export((int)$orderId);
        } catch (\Throwable $e) {
            $this->logger->error('Order export failed', [
                'order_id' => $orderId,
                'exception' => $e,
            ]);
            $this->exporter->markForRetry((int)$orderId);
        }
    }
}

The catch block matters. With the default consumer class, when the handler throws, the message is rejected and is not put back on the queue. If you do not record the failure somewhere, it is simply gone. Logging plus a retry flag on your own table gives you something to act on.

How consumers are started

Consumers are long running PHP processes started with bin/magento queue:consumers:start. There are two ways to keep them alive: let Magento cron spawn them, or run them under a process manager.

The cron consumers runner

By default the cron job consumers_runner runs every minute and starts any consumer that is not already running. It is controlled by the cron_consumers_runner section of env.php:

'cron_consumers_runner' => [
    'cron_run' => true,
    'max_messages' => 10000,
    'consumers' => [
        'vendor.order.export',
        'async.operations.all',
        'product_action_attribute.update',
    ],
],
'queue' => [
    'consumers_wait_for_messages' => 1,
],

• cron_run turns the runner on or off.
• max_messages is how many messages a consumer handles before it exits. The next cron run starts a fresh one. This caps memory growth and makes the process pick up new code after a deploy.
• consumers limits which consumers cron starts. An empty array means all of them. Listing them explicitly avoids starting dozens of processes you do not use.
• consumers_wait_for_messages set to 1 keeps a consumer waiting for new messages. Set to 0, a consumer exits as soon as the queue is empty, which saves memory on small servers at the cost of up to a minute of delay.

The weak point is obvious once you see it: this only works while Magento cron runs. If the cron line is missing from the crontab after a server move, or the cron process is stuck on a lock, no consumer is restarted after it exits. Nothing on the storefront tells you.

Supervisor or systemd

For consumers that matter to the business, a process manager is more predictable. It restarts the process when it exits for any reason, and you can see its state with one command. Disable cron for those consumers, either with cron_run set to false or by leaving them out of the consumers list, so you never have two owners for the same process.

[program:magento-order-export]
command=/usr/bin/php /var/www/shop/current/bin/magento queue:consumers:start vendor.order.export --single-thread --max-messages=1000
directory=/var/www/shop/current
user=www-data
numprocs=1
autostart=true
autorestart=true
startsecs=5
stopwaitsecs=60
redirect_stderr=true
stdout_logfile=/var/log/supervisor/magento-order-export.log

The same with systemd, as a template unit so one file can serve several consumers:

# /etc/systemd/system/[email protected]
[Unit]
Description=Magento consumer %i
After=network.target

[Service]
User=www-data
WorkingDirectory=/var/www/shop/current
ExecStart=/usr/bin/php bin/magento queue:consumers:start %i --single-thread --max-messages=1000
Restart=always
RestartSec=5

[Install]
WantedBy=multi-user.target

systemctl enable --now [email protected]
systemctl status [email protected]

Keep --max-messages even under a process manager. A clean exit and restart every few thousand messages is the simplest protection against memory leaks, stale database connections and old code still loaded after a release.

Why consumers die silently

These are the causes worth checking first, roughly in the order they show up in practice.

• Cron is not running. The consumer exited after max_messages and nothing started it again. Check crontab -l for the Magento user and the cron_schedule table for recent consumers_runner entries.
• The handler throws. The message is rejected and lost. Look in var/log/system.log and var/log/exception.log.
• Fatal errors. A memory limit or a type error kills the PHP process. Without --max-messages and a process manager, it stays dead.
• Database connection lost. A consumer that waits a long time can hit the MySQL wait_timeout. Restarting after a fixed number of messages reduces the window.
• Old code after a deploy. A running consumer keeps the classes it loaded at start. Restart all consumers as part of every release.
• Broken connection settings. Wrong credentials or virtual host in env.php, or a firewall change. The consumer fails at start every time it is launched and only the logs show it.

Monitoring the queue, not the process

A running process does not prove that messages are flowing. The useful signal is queue depth and the number of consumers attached to each queue. On the RabbitMQ side:

rabbitmqctl list_queues -p / name messages messages_unacknowledged consumers

Alert when a queue you care about has zero consumers, or when its message count keeps growing for longer than your normal processing time. The RabbitMQ management plugin exposes the same numbers over HTTP if your monitoring prefers that.

On the Magento side, ps aux | grep queue:consumers:start shows what is running, and the queue_lock table holds message locks if you need to understand duplicate processing.

Deploy checklist

• Run setup:upgrade so new exchanges, queues and bindings are created.
• Restart every consumer after the code switch, with supervisorctl restart or systemctl restart.
• If cron owns the consumers, kill the old processes and let the next cron run start them on the new code.
• Check queue depth a few minutes later.

FAQ

Do I need RabbitMQ to use message queues in Magento 2?

No. Magento Open Source can use MySQL as the queue backend for many topics. RabbitMQ is recommended for production because it handles volume better and some features expect an AMQP connection.

What does max-messages do in queue:consumers:start?

It sets how many messages the consumer processes before it exits. Combined with cron or a process manager that restarts it, this limits memory growth and makes sure new code is loaded after deploys.

Should I use cron_consumers_runner or supervisor?

Cron is fine for low priority consumers. For consumers the business depends on, supervisor or systemd restarts them immediately and makes their state visible. Do not let both manage the same consumer.

Why are messages piling up in RabbitMQ with no errors in Magento?

Usually no consumer is attached to the queue. Check the consumers column of rabbitmqctl list_queues, then check that Magento cron runs or that your process manager is up.

I am Rodrigo Biazzi, a Magento Certified Developer working with Magento and Adobe Commerce for more than 12 years. I am available for freelance Magento work, from single fixes to longer projects. If this is the kind of problem you have on your store, get in touch.