Custom Message Queue Consumers in Magento 2: Retry Logic, Dead Letter Handling, Monitoring

1. Context: why custom consumers are necessary

The standard RabbitMQ setup in Magento 2 explains how an exchange, a queue and a binding relate to each other. These basics are a prerequisite, but they do not answer the real question in production: what happens when processing a message fails. A message queue consumer that simply accepts and processes messages is fine for a demo. The moment an external API call times out, a database connection briefly drops, or a message carries a malformed payload, it becomes clear whether the consumer was built to be robust or silently loses data.

The core of the problem is idempotency. An MQ consumer must assume that any message can arrive more than once, whether through a requeue after a failure, a restart of the consumer process mid-processing, or a manual replay from the dead letter queue. Anyone who ignores this case produces duplicate order exports, duplicate emails, or duplicate bookings in an external system. Service Contracts and repositories are not just an architectural principle from the Magento style guide here, they are the tool that reliably checks the current state of an entity before a potentially expensive operation runs again.

In this article we build a custom message queue consumer for a realistic scenario: exporting orders to an external ERP system over a topic called order.export. Using this example, we cover architecture, idempotency checks, retry logic with backoff, dead letter handling and monitoring, each with code that carries over directly to other topics.

2. Consumer architecture: topology, communication and declaration

A custom message queue consumer in Magento 2 consists of four interlocking declarations. communication.xml defines the topic and the schema class of the message. queue_topology.xml defines the exchange and the binding to the queue. queue_publisher.xml connects the topic to a connection and an exchange, so a publisher knows where to send a message at all. Finally, queue_consumer.xml registers the consumer class that reads from the queue, including the maximum number of messages per run.

At first glance these four files look redundant, but they deliberately separate concerns: topology is infrastructure, communication is the contract, publisher is sender configuration, consumer is receiver configuration. This separation makes it possible to serve the same contract from multiple publishers, or to bind multiple consumer instances to the same queue, without touching the consumer class code itself. The example below shows a minimal but complete declaration for the order.export topic.

<!-- app/code/Mironsoft/OrderExport/etc/communication.xml -->
<config xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:noNamespaceSchemaLocation="urn:magento:framework:Communication/etc/communication.xsd">
    <topic name="order.export" schema="Mironsoft\OrderExport\Api\Data\OrderExportMessageInterface"/>
</config>

<!-- app/code/Mironsoft/OrderExport/etc/queue_topology.xml -->
<config xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:noNamespaceSchemaLocation="urn:magento:framework-message-queue:etc/topology.xsd">
    <exchange name="mironsoft.order.export" type="topic" connection="amqp">
        <binding id="orderExportBinding" topic="order.export"
                 destinationType="queue" destination="order.export.queue"/>
    </exchange>
</config>

<!-- app/code/Mironsoft/OrderExport/etc/queue_publisher.xml -->
<config xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:noNamespaceSchemaLocation="urn:magento:framework:MessageQueue/etc/publisher.xsd">
    <publisher topic="order.export">
        <connection name="amqp" exchange="mironsoft.order.export"/>
    </publisher>
</config>

<!-- app/code/Mironsoft/OrderExport/etc/queue_consumer.xml -->
<config xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:noNamespaceSchemaLocation="urn:magento:framework:MessageQueue/etc/consumer.xsd">
    <consumer name="order.export.consumer" queue="order.export.queue" connection="amqp"
              consumerInstance="Mironsoft\OrderExport\Model\Consumer\OrderExportConsumer"
              maxMessages="500"/>
</config>

After changing any of these four files, the consumer must be restarted, because topology and bindings are only read at startup and are not re-evaluated on every message. In practice this means: stop bin/magento queue:consumers:start order.export.consumer, clear the cache, and restart the consumer process, ideally through the systemd service from section 6.

3. The consumer class itself

The actual consumer class is a plain PHP class with a process method that Magento references through consumerInstance. With PHP 8.4 we consistently use constructor property promotion for all injected dependencies: our own status repository for the idempotency check, an API client for the external system call, and PublisherInterface to route to a retry or dead letter queue on failure. We deliberately avoid preferences here, instead every extension point is realized through plugins in case other modules need to hook into the export.

The most important part of this class is the idempotency check right at the start of the process method. Before any expensive operation runs, the message queue consumer queries a dedicated status flag maintained either as its own entity or as an extra column on the order. If the order was already exported successfully, the message is acknowledged without further action. Only then does the actual processing happen, wrapped in a try-catch around the external call so a failure can be handed off to the retry logic in a controlled way, instead of letting the exception fall out of the process method unhandled.

<?php

declare(strict_types=1);

namespace Mironsoft\OrderExport\Model\Consumer;

use Magento\Framework\Exception\LocalizedException;
use Magento\Framework\MessageQueue\PublisherInterface;
use Mironsoft\OrderExport\Api\Data\OrderExportMessageInterface;
use Mironsoft\OrderExport\Api\OrderExportStatusRepositoryInterface;
use Mironsoft\OrderExport\Model\ExternalApiClient;
use Psr\Log\LoggerInterface;

/**
 * Consumes order.export messages and forwards order data to an external ERP system.
 */
class OrderExportConsumer
{
    private const MAX_RETRIES = 5;

    /**
     * @param OrderExportStatusRepositoryInterface $statusRepository Repository for the idempotency flag entity.
     * @param ExternalApiClient $apiClient Client for the external ERP endpoint.
     * @param PublisherInterface $publisher Used to route failed messages to retry or dead letter queues.
     * @param LoggerInterface $logger Dedicated logger channel for order export events.
     */
    public function __construct(
        private readonly OrderExportStatusRepositoryInterface $statusRepository,
        private readonly ExternalApiClient $apiClient,
        private readonly PublisherInterface $publisher,
        private readonly LoggerInterface $logger
    ) {
    }

    /**
     * Processes a single order.export message with an idempotency guard.
     *
     * @param OrderExportMessageInterface $message
     * @return void
     */
    public function process(OrderExportMessageInterface $message): void
    {
        $orderId = $message->getOrderId();
        $status = $this->statusRepository->getByOrderId($orderId);

        // Idempotency guard: skip if this order was already exported successfully.
        if ($status->getExported()) {
            $this->logger->info(sprintf('Order %d already exported, skipping.', $orderId));
            return;
        }

        try {
            $this->apiClient->sendOrder($orderId);
            $status->setExported(true);
            $status->setLastError(null);
            $this->statusRepository->save($status);
        } catch (LocalizedException $exception) {
            $this->handleFailure($message, $status, $exception);
        }
    }

    /**
     * Increments the retry counter and either rethrows for a requeue or routes to the dead letter queue.
     *
     * @param OrderExportMessageInterface $message
     * @param \Mironsoft\OrderExport\Api\Data\OrderExportStatusInterface $status
     * @param LocalizedException $exception
     * @return void
     * @throws LocalizedException
     */
    private function handleFailure(
        OrderExportMessageInterface $message,
        $status,
        LocalizedException $exception
    ): void {
        $retryCount = $status->getRetryCount() + 1;
        $status->setRetryCount($retryCount);
        $status->setLastError($exception->getMessage());
        $this->statusRepository->save($status);

        if ($retryCount >= self::MAX_RETRIES) {
            $this->logger->error(sprintf(
                'Order %d exceeded max retries, routing to dead letter queue.',
                $message->getOrderId()
            ));
            $this->publisher->publish('order.export.deadletter', $message);
            return;
        }

        $this->logger->warning(sprintf(
            'Order %d export failed (attempt %d), will be retried.',
            $message->getOrderId(),
            $retryCount
        ));
        throw $exception;
    }
}

4. Retry logic and exponential backoff

Magento does offer a max_messages attribute in queue_consumer.xml, but no native backoff strategy for failed messages. Simply rethrowing an exception does make RabbitMQ requeue the message, but without any delay. The result: a broken external service gets hit again every second, which makes the situation worse rather than better. A serious MQ consumer therefore needs a self-implemented retry counter and exponential backoff that extends the wait time between attempts with every failure.

Important detail: sleep() inside the consumer process is not a good solution, because it blocks the worker for the entire wait time and no other messages can be processed. The better solution is a delayed exchange, as provided by the RabbitMQ plugin rabbitmq_delayed_message_exchange. The message is republished with an x-delay header, RabbitMQ holds it back for the specified time, and then automatically delivers it to the original queue. The consumer process stays free for other messages in the meantime.

<?php

declare(strict_types=1);

namespace Mironsoft\OrderExport\Model\Consumer;

use Magento\Framework\MessageQueue\PublisherInterface;

/**
 * Requeues failed order.export messages with an exponential backoff delay
 * instead of blocking the consumer process with sleep().
 */
class RetryScheduler
{
    private const BASE_DELAY_MS = 2000;
    private const MAX_DELAY_MS = 300000;
    private const MAX_RETRIES = 5;

    /**
     * @param PublisherInterface $publisher Publisher used to send the message to a delayed retry queue.
     */
    public function __construct(
        private readonly PublisherInterface $publisher
    ) {
    }

    /**
     * Schedules a retry for the given message via a delayed exchange.
     *
     * @param string $orderExportPayload Serialized message payload.
     * @param int $retryCount Number of attempts already made.
     * @return bool True if a retry was scheduled, false if retries are exhausted.
     */
    public function scheduleRetry(string $orderExportPayload, int $retryCount): bool
    {
        if ($retryCount >= self::MAX_RETRIES) {
            return false;
        }

        $delayMs = min(self::MAX_DELAY_MS, self::BASE_DELAY_MS * (2 ** $retryCount));

        // x-delay is read by the rabbitmq_delayed_message_exchange plugin
        // and routes the message back to order.export.queue after the delay.
        $this->publisher->publish('order.export.retry', $orderExportPayload, [
            'x-delay' => $delayMs,
            'x-retry-count' => $retryCount + 1,
        ]);

        return true;
    }
}

With this strategy, the first retry waits 2 seconds, the second 4 seconds, the third 8 seconds, and so on up to a cap of 5 minutes. A jitter of a few hundred milliseconds additionally prevents many failed messages from being redelivered at exactly the same moment and immediately overwhelming an external service that has just come back online.

5. Dead letter handling

When a message queue consumer has reached the maximum number of retries, the message must not simply be dropped. It has to move into a dedicated dead letter queue where it can later be inspected, analyzed and, if needed, manually replayed. In Magento 2 this means setting up an additional exchange and queue, either through a native RabbitMQ dead-letter-exchange argument on the binding, or, as shown in the consumer example above, through an explicit publish call once retries are exhausted.

The native approach via x-dead-letter-exchange has the advantage that RabbitMQ itself takes over the forwarding as soon as a message is negatively acknowledged (nack) or its TTL expires, without the consumer code having to trigger it explicitly. For this, queue_topology.xml extends the binding of the main queue with an argument pointing to a separate fanout exchange, which in turn is bound to the dead letter queue.

<!-- app/code/Mironsoft/OrderExport/etc/queue_topology.xml -->
<config xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:noNamespaceSchemaLocation="urn:magento:framework-message-queue:etc/topology.xsd">

    <!-- Dead letter exchange and queue, bound as a fanout target -->
    <exchange name="mironsoft.order.export.dlx" type="fanout" connection="amqp">
        <binding id="orderExportDeadLetterBinding" topic=""
                 destinationType="queue" destination="order.export.deadletter.queue"/>
    </exchange>

    <!-- Main exchange, routes rejected messages to the dead letter exchange -->
    <exchange name="mironsoft.order.export" type="topic" connection="amqp">
        <binding id="orderExportBinding" topic="order.export"
                 destinationType="queue" destination="order.export.queue">
            <arguments>
                <argument name="x-dead-letter-exchange" xsi:type="string">mironsoft.order.export.dlx</argument>
            </arguments>
        </binding>
    </exchange>
</config>

Two commands are enough to inspect the dead letter queue, and both can be wired into a daily report: bin/magento queue:consumers:list shows all registered consumers along with their queues, while rabbitmqctl list_queues name messages consumers shows the current depth of every queue directly, including the dead letter queue. A growing dead letter queue is a reliable early warning that an external service has become unstable or that a new data shape no longer matches the expected schema.

6. Monitoring in production

An MQ consumer started manually in a terminal via bin/magento queue:consumers:start survives neither a server restart nor a crash. In production, every consumer belongs in its own systemd service with Restart=always, so a crashed process restarts automatically without anyone having to step in manually. Separate units per topic also allow restarting a single consumer in a targeted way, for example after a change to its consumer class, without stopping every other consumer.

A running process alone is not enough for real monitoring. What matters is queue depth: if messages arrive faster than the consumer can process them, the queue keeps growing even though the process itself is technically running. A simple cron script that queries queue depth via rabbitmqctl and alerts once a threshold is exceeded covers most failure scenarios without needing a full APM tool.

# /etc/systemd/system/magento-consumer-order-export.service
[Unit]
Description=Magento 2 MQ Consumer: order.export
After=network.target mysql.service rabbitmq-server.service

[Service]
Type=simple
User=magento
WorkingDirectory=/var/www/html
ExecStart=/usr/bin/php bin/magento queue:consumers:start order.export.consumer --max-messages=10000
Restart=always
RestartSec=5
StandardOutput=append:/var/log/magento/consumer-order-export.log
StandardError=append:/var/log/magento/consumer-order-export-error.log

[Install]
WantedBy=multi-user.target

# Enable and start the service
sudo systemctl daemon-reload
sudo systemctl enable --now magento-consumer-order-export.service
sudo systemctl status magento-consumer-order-export.service

# List all registered consumers and their queue bindings
bin/magento queue:consumers:list

# Inspect queue depth and active consumers directly via RabbitMQ
rabbitmqctl list_queues name messages consumers -p /mironsoft

# Simple alerting: notify if queue depth exceeds a threshold
depth=$(rabbitmqctl list_queues name messages -p /mironsoft | grep order.export.queue | awk '{print $2}')
if [ "$depth" -gt 5000 ]; then
  echo "ALERT: order.export.queue depth is $depth" | mail -s "Queue Alert" ops@mironsoft.de
fi

7. Scaling MQ consumers

A single consumer process handles messages sequentially, one after another. Once throughput is no longer sufficient, the obvious first step is to run multiple instances of the same consumer in parallel, either as several systemd units with different instance names or through a systemd template unit using the @ syntax. RabbitMQ then distributes messages round-robin across all connected consumer instances of the same queue, as long as the prefetch count is set low enough that no instance is permanently overloaded while others sit idle.

The prefetch count determines how many unacknowledged messages RabbitMQ delivers to a consumer connection at once. Too high a value means a slow instance holds many messages while other instances have already finished and are waiting for new work. A value of 1 gives the fairest distribution but costs some throughput due to the extra overhead per acknowledgment. In practice, a value between 5 and 20 is a good compromise for most topics.

Scaling pays off mainly when queue depth keeps growing despite a healthily running single consumer, not at every short-lived spike. Anyone who scales prematurely to ten parallel instances often just shifts the problem onto the external service, which then faces ten times the load at once. A moderate increase, combined with monitoring queue depth over time, is the more reliable way to find the right number of consumer instances.

8. Common mistakes with custom consumers

The most common mistake with a self-built message queue consumer is missing or incorrect acknowledgment logic. If a message is acknowledged before processing has actually completed successfully, for example because an exception in a downstream code path is not caught, RabbitMQ considers the message done and it is irrevocably lost. Conversely, a complete absence of error handling means every exception crashes the entire consumer process, which without a systemd restart means processing stops permanently.

A second classic mistake is blocking business logic directly inside the consumer, such as a synchronous HTTP call without a timeout. If the external service hangs, the entire consumer process hangs too, and every subsequent message piles up in the queue even though the process itself is shown as running. Every external call inside an MQ consumer therefore needs an explicit timeout that is noticeably shorter than the default timeout of the HTTP client used.

A third mistake concerns missing idempotency combined with requeue logic: without the status check shown in section 3, every retry attempt re-runs the entire operation, including the parts that already succeeded. For an order export this can mean duplicate bookings in the target system in the worst case, which then have to be cleaned up manually. Together, these three mistakes explain most production incidents around self-built consumers.

9. Retry and DLQ strategies compared

There are several ways to implement retry logic and dead letter handling for a message queue consumer, with clear differences in reliability and operational behavior. The table below compares the common approaches directly.

Overall, the comparison shows that almost every unsafe approach rests on the same underlying problem: error handling is left to the default infrastructure instead of being modeled explicitly inside the message queue consumer itself. Combining the five recommended patterns consistently produces a consumer that survives outages instead of amplifying them.

10. Summary

A production-ready message queue consumer in Magento 2 differs from the standard RabbitMQ tutorial mainly in what is missing between the lines of the official documentation: an idempotency check via a dedicated status flag, non-blocking retry with exponential backoff via a delayed exchange, an explicit dead letter queue for exhausted retries, and systemd-based monitoring that keeps an eye on both process state and queue depth.

None of these elements is complicated on its own, but how they work together decides whether an outage of an external system results in a handful of delayed messages or in data loss and manual rework. Anyone building new topics following the same pattern as order.export in this article, with clearly separated topology, a lean consumer class, and explicit retry and DLQ logic, significantly reduces the operational effort for every additional MQ consumer in the project.

11. FAQ: Message Queue Consumers in Magento 2