Eigene Message Queue Consumer in Magento 2: Retry-Logik, Dead-Letter-Handling, Monitoring

1. Einordnung: Warum eigene Consumer noetig sind

Das Standard-RabbitMQ-Setup in Magento 2 erklaert, wie ein Exchange, eine Queue und ein Binding zusammenhaengen. Diese Grundlagen sind Voraussetzung, aber sie beantworten nicht die eigentliche Frage im produktiven Betrieb: Was passiert, wenn die Verarbeitung einer Nachricht fehlschlaegt. Ein Message Queue Consumer, der einfach nur Nachrichten entgegennimmt und verarbeitet, ist fuer Demo-Zwecke ausreichend. Sobald ein externer API-Aufruf timeout, eine Datenbank-Verbindung kurzzeitig weg ist oder eine Nachricht fehlerhafte Payload-Daten enthaelt, zeigt sich, ob der Consumer robust gebaut wurde oder Daten stillschweigend verliert.

Der Kern des Problems ist Idempotenz. Ein MQ-Consumer muss davon ausgehen, dass jede Nachricht mehr als einmal ankommen kann, sei es durch ein Requeue nach einem Fehler, durch einen Neustart des Consumer-Prozesses waehrend der Verarbeitung oder durch ein manuelles Replay aus der Dead-Letter-Queue. Wer diesen Fall nicht behandelt, produziert doppelte Bestellexporte, doppelte E-Mails oder doppelte Buchungen in einem externen System. Service Contracts und Repositories sind hier nicht nur ein Architekturprinzip aus dem Magento-Styleguide, sondern das Werkzeug, mit dem man den aktuellen Zustand einer Entity zuverlaessig abfragt, bevor man eine potenziell teure Operation erneut ausfuehrt.

In diesem Artikel bauen wir einen eigenen Message Queue Consumer fuer ein realistisches Szenario: den Export von Bestellungen an ein externes ERP-System ueber ein Topic namens order.export. Anhand dieses Beispiels zeigen wir Architektur, Idempotenz-Pruefung, Retry-Logik mit Backoff, Dead-Letter-Handling und Monitoring, jeweils mit Code, der sich direkt auf andere Topics uebertragen laesst.

2. Consumer-Architektur: Topologie, Kommunikation und Deklaration

Ein eigener Message Queue Consumer besteht in Magento 2 aus vier zusammenspielenden Deklarationen. Die communication.xml definiert das Topic und die Schema-Klasse der Nachricht. Die queue_topology.xml definiert den Exchange und das Binding zur Queue. Die queue_publisher.xml verbindet das Topic mit einer Connection und einem Exchange, damit ein Publisher ueberhaupt weiss, wohin er eine Nachricht schickt. Die queue_consumer.xml schliesslich registriert die Consumer-Klasse, die aus der Queue liest, inklusive der maximalen Anzahl an Nachrichten pro Lauf.

Diese vier Dateien wirken auf den ersten Blick redundant, trennen aber bewusst Verantwortlichkeiten: Topologie ist Infrastruktur, Communication ist Vertrag, Publisher ist Absender-Konfiguration, Consumer ist Empfaenger-Konfiguration. Diese Trennung erlaubt es, denselben Vertrag von mehreren Publishern aus zu bedienen oder mehrere Consumer-Instanzen an dieselbe Queue zu binden, ohne den Code der Consumer-Klasse selbst anzufassen. Das folgende Beispiel zeigt eine minimale, aber vollstaendige Deklaration fuer das Topic order.export.

<!-- 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>

Nach einer Aenderung an diesen vier Dateien muss der Consumer neu gestartet werden, denn Topologie und Bindings werden nur beim Start eingelesen und nicht bei jedem Nachrichten-Empfang neu ausgewertet. In der Praxis bedeutet das: bin/magento queue:consumers:start order.export.consumer beenden, Cache leeren und den Consumer-Prozess neu starten, idealerweise ueber den systemd-Service aus Abschnitt 6.

3. Die Consumer-Klasse selbst

Die eigentliche Consumer-Klasse ist eine einfache PHP-Klasse mit einer process-Methode, die von Magento ueber consumerInstance referenziert wird. Mit PHP 8.4 nutzen wir Constructor Property Promotion konsequent fuer alle injizierten Abhaengigkeiten: die eigene Status-Repository fuer die Idempotenz-Pruefung, einen API-Client fuer den externen Systemaufruf und den PublisherInterface, um im Fehlerfall an eine Retry- oder Dead-Letter-Queue weiterzuleiten. Auf Preferences verzichten wir bewusst, stattdessen wird jede Erweiterung ueber Plugins realisiert, falls andere Module in den Export eingreifen muessen.

Der wichtigste Teil dieser Klasse ist die Idempotenz-Pruefung ganz am Anfang der process-Methode. Bevor irgendeine teure Operation ausgefuehrt wird, fragt der Message Queue Consumer ein eigenes Status-Flag ab, das in einer eigenen Entity oder als zusaetzliche Spalte an der Order gepflegt wird. Ist die Order bereits erfolgreich exportiert, wird die Nachricht ohne weitere Aktion bestaetigt. Erst danach folgt die eigentliche Verarbeitung mit Try-Catch um den externen Aufruf, damit ein Fehler kontrolliert in die Retry-Logik uebergeben werden kann, statt die Exception unbehandelt aus der process-Methode fallen zu lassen.

<?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-Logik und exponentielles Backoff

Magento liefert in queue_consumer.xml zwar ein max_messages-Attribut, aber keine native Backoff-Strategie fuer fehlgeschlagene Nachrichten. Wer eine Exception einfach weiterwirft, sorgt zwar dafuer, dass RabbitMQ die Nachricht requeued, aber ohne Verzoegerung. Das Ergebnis: ein fehlerhafter externer Service wird im Sekundentakt erneut angefragt, was die Situation eher verschlimmert als verbessert. Ein serioeser MQ-Consumer braucht daher einen selbst implementierten Retry-Counter und ein exponentielles Backoff, das die Wartezeit zwischen den Versuchen mit jedem Fehlschlag verlaengert.

Wichtig dabei: sleep() im Consumer-Prozess ist keine gute Loesung, weil es den Worker fuer die komplette Wartezeit blockiert und keine anderen Nachrichten mehr verarbeitet werden koennen. Die bessere Loesung ist eine Delayed Exchange, wie sie das RabbitMQ-Plugin rabbitmq_delayed_message_exchange bereitstellt. Die Nachricht wird mit einem x-delay Header erneut publiziert, RabbitMQ haelt sie fuer die angegebene Zeit zurueck und liefert sie danach automatisch an die urspruengliche Queue aus. Der Consumer-Prozess bleibt waehrenddessen frei fuer andere Nachrichten.

<?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;
    }
}

Mit dieser Strategie wartet die erste Wiederholung 2 Sekunden, die zweite 4 Sekunden, die dritte 8 Sekunden und so weiter, bis eine Obergrenze von 5 Minuten erreicht ist. Ein Jitter von einigen hundert Millisekunden verhindert zusaetzlich, dass viele fehlgeschlagene Nachrichten exakt gleichzeitig erneut zugestellt werden und den gerade erst wieder erreichbaren externen Service sofort wieder ueberlasten.

5. Dead-Letter-Handling

Wenn ein Message Queue Consumer die maximale Anzahl an Retries erreicht hat, darf die Nachricht nicht einfach verworfen werden. Sie muss in eine eigene Dead-Letter-Queue wandern, aus der sie spaeter inspiziert, analysiert und bei Bedarf manuell erneut eingespielt werden kann. In Magento 2 richtet man dafuer einen zusaetzlichen Exchange und eine zusaetzliche Queue ein, entweder ueber ein natives RabbitMQ dead-letter-exchange Argument am Binding oder, wie im Beispiel des Consumers oben gezeigt, ueber eine explizite publish-Anweisung nach Erschoepfung der Retries.

Der native Ansatz ueber x-dead-letter-exchange hat den Vorteil, dass RabbitMQ selbst die Weiterleitung uebernimmt, sobald eine Nachricht negativ bestaetigt wird (nack) oder ihre TTL ablaeuft, ohne dass der Consumer-Code das explizit anstossen muss. Die queue_topology.xml erweitert dafuer das Binding der Haupt-Queue um ein Argument, das auf einen separaten Fanout-Exchange verweist, an den wiederum die Dead-Letter-Queue gebunden ist.

<!-- 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>

Zur Inspektion der Dead-Letter-Queue reichen zwei Befehle aus, die man auch in ein taegliches Reporting einbauen kann: bin/magento queue:consumers:list zeigt alle registrierten Consumer inklusive ihrer Queues, waehrend rabbitmqctl list_queues name messages consumers direkt die aktuelle Tiefe jeder Queue anzeigt, inklusive der Dead-Letter-Queue. Eine wachsende Dead-Letter-Queue ist ein zuverlaessiges Fruehwarnsignal dafuer, dass ein externer Service instabil geworden ist oder eine neue Datenstruktur nicht mehr zum erwarteten Schema passt.

6. Monitoring im Produktionsbetrieb

Ein MQ-Consumer, der per bin/magento queue:consumers:start manuell in einem Terminal gestartet wird, ueberlebt keinen Server-Neustart und keinen Absturz. Im Produktionsbetrieb gehoert jeder Consumer in einen eigenen systemd-Service mit Restart=always, damit ein abgestuerzter Prozess automatisch neu startet, ohne dass jemand manuell eingreifen muss. Getrennte Units pro Topic erlauben zudem, einen einzelnen Consumer gezielt neu zu starten, etwa nach einer Aenderung an der zugehoerigen Consumer-Klasse, ohne alle anderen Consumer zu stoppen.

Fuer echtes Monitoring reicht ein laufender Prozess allein nicht aus. Entscheidend ist die Queue-Tiefe: Wenn Nachrichten schneller ankommen als der Consumer sie verarbeiten kann, waechst die Queue kontinuierlich an, auch wenn der Prozess selbst technisch laeuft. Ein einfaches Cronjob-Skript, das die Queue-Tiefe per rabbitmqctl abfragt und bei Ueberschreiten eines Schwellwerts alarmiert, deckt die meisten Ausfallszenarien ab, ohne dass ein vollstaendiges APM-Tool noetig waere.

# /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. Skalierung von MQ-Consumern

Ein einzelner Consumer-Prozess verarbeitet Nachrichten sequenziell, eine nach der anderen. Sobald der Durchsatz nicht mehr ausreicht, ist der naheliegende erste Schritt, mehrere Instanzen desselben Consumers parallel zu starten, entweder als mehrere systemd-Units mit unterschiedlichen Instanznamen oder ueber einen systemd-Template-Unit mit @-Syntax. RabbitMQ verteilt Nachrichten dann per Round-Robin an alle verbundenen Consumer-Instanzen derselben Queue, solange der Prefetch-Count klein genug eingestellt ist, dass keine Instanz staendig ueberladen wird, waehrend andere leerlaufen.

Der Prefetch-Count bestimmt, wie viele unbestaetigte Nachrichten RabbitMQ gleichzeitig an eine Consumer-Verbindung ausliefert. Ein zu hoher Wert fuehrt dazu, dass eine langsame Instanz viele Nachrichten haelt, waehrend andere Instanzen bereits fertig sind und auf neue Arbeit warten muessten. Ein Wert von 1 sorgt fuer die fairste Verteilung, kostet aber etwas Durchsatz durch den zusaetzlichen Overhead pro Bestaetigung. In der Praxis ist ein Wert zwischen 5 und 20 fuer die meisten Topics ein guter Kompromiss.

Skalierung lohnt sich vor allem dann, wenn die Queue-Tiefe trotz eines gesund laufenden einzelnen Consumers kontinuierlich waechst, nicht schon bei jedem kurzfristigen Anstieg. Wer vorschnell auf zehn parallele Instanzen skaliert, verlagert das Problem oft nur auf den externen Service, der dann mit zehnfacher Last gleichzeitig konfrontiert wird. Eine moderate Erhoehung, kombiniert mit Monitoring der Queue-Tiefe ueber die Zeit, ist der zuverlaessigere Weg, die richtige Anzahl an Consumer-Instanzen zu ermitteln.

8. Typische Fehler bei eigenen Consumern

Der haeufigste Fehler bei einem selbst gebauten Message Queue Consumer ist fehlende oder falsche Acknowledgment-Logik. Wird eine Nachricht bestaetigt, bevor die eigentliche Verarbeitung erfolgreich abgeschlossen ist, etwa weil eine Exception in einem nachgelagerten Codepfad nicht abgefangen wird, gilt die Nachricht fuer RabbitMQ als erledigt und ist unwiderruflich verloren. Umgekehrt fuehrt das komplette Fehlen von Fehlerbehandlung dazu, dass jede Exception den gesamten Consumer-Prozess zum Absturz bringt, was ohne systemd-Restart eine dauerhaft stillstehende Verarbeitung bedeutet.

Ein zweiter klassischer Fehler ist blockierende Business-Logik direkt im Consumer, etwa ein synchroner HTTP-Aufruf ohne Timeout. Haengt der externe Service, haengt auch der komplette Consumer-Prozess, und alle nachfolgenden Nachrichten stauen sich in der Queue, obwohl der Prozess selbst als laufend angezeigt wird. Jeder externe Aufruf innerhalb eines MQ-Consumers braucht daher einen expliziten Timeout, der deutlich kuerzer ist als das Standard-Timeout des verwendeten HTTP-Clients.

Ein dritter Fehler betrifft fehlende Idempotenz in Kombination mit Requeue-Logik: Ohne die in Abschnitt 3 gezeigte Status-Pruefung fuehrt jeder Retry-Versuch die komplette Operation erneut aus, auch die Teile, die bereits erfolgreich waren. Bei einem Bestellexport bedeutet das im schlimmsten Fall doppelte Buchungen im Zielsystem, die anschliessend manuell bereinigt werden muessen. Diese drei Fehler zusammen erklaeren die meisten produktiven Vorfaelle rund um selbst entwickelte Consumer.

9. Retry- und DLQ-Strategien im Vergleich

Es gibt mehrere Wege, Retry-Logik und Dead-Letter-Handling fuer einen Message Queue Consumer umzusetzen, mit deutlichen Unterschieden bei Zuverlaessigkeit und Betriebsverhalten. Die folgende Uebersicht zeigt die gaengigen Ansaetze im direkten Vergleich.

In der Summe zeigt der Vergleich, dass fast jeder unsichere Ansatz auf demselben Grundproblem beruht: Fehlerbehandlung wird der Standard-Infrastruktur ueberlassen, statt sie explizit im eigenen Message Queue Consumer zu modellieren. Wer die fuenf empfohlenen Pattern konsequent kombiniert, erhaelt einen Consumer, der Ausfaelle uebersteht, statt sie zu verstaerken.

10. Zusammenfassung

Ein produktionsreifer Message Queue Consumer in Magento 2 unterscheidet sich vom Standard-RabbitMQ-Tutorial vor allem durch das, was zwischen den Zeilen der offiziellen Dokumentation fehlt: Idempotenz-Pruefung ueber ein eigenes Status-Flag, nicht-blockierendes Retry mit exponentiellem Backoff ueber eine Delayed Exchange, eine explizite Dead-Letter-Queue fuer erschoepfte Retries und systemd-basiertes Monitoring, das sowohl den Prozesszustand als auch die Queue-Tiefe im Blick behaelt.

Keines dieser Elemente ist fuer sich genommen kompliziert, aber ihr Zusammenspiel entscheidet darueber, ob ein Ausfall eines externen Systems zu ein paar verzoegerten Nachrichten fuehrt oder zu Datenverlust und manueller Nacharbeit. Wer neue Topics nach demselben Muster wie order.export in diesem Artikel aufbaut, also mit klar getrennter Topologie, einer schlanken Consumer-Klasse und expliziter Retry- und DLQ-Logik, reduziert den operativen Aufwand fuer jeden weiteren MQ-Consumer im Projekt erheblich.

11. FAQ: Message Queue Consumer in Magento 2