Symfony Messenger: Eigene Bus-Middleware für Cross-Cutting-Concerns schreiben

1. Was Middleware im Messenger-Bus eigentlich macht

Jede Message, die über $bus->dispatch() verschickt wird, durchläuft nicht direkt den zuständigen Handler, sondern eine Kette von Middleware-Objekten, die nacheinander aufgerufen werden. Jede Middleware bekommt das aktuelle Envelope, das die Message selbst sowie beliebige Metadaten in Form von Stamps enthält, und einen Stack-Parameter, über den sie explizit die nächste Middleware in der Kette aufruft. Erst die letzte Middleware in der Kette ruft tatsächlich den registrierten Handler auf, sodass jede vorherige Middleware Code sowohl vor als auch nach der eigentlichen Verarbeitung ausführen kann.

Dieses Muster ist bewusst an HTTP-Middleware aus anderen Frameworks angelehnt, denn es löst dasselbe strukturelle Problem: Cross-Cutting-Concerns, die für praktisch jede Message relevant sind, sollen an einer zentralen Stelle implementiert werden, statt in jedem Handler wiederholt zu werden. Symfony liefert bereits eine Reihe eingebauter Middleware mit, etwa für Doctrine-Transaktionen oder Fehlerbehandlung, und eigene Middleware reiht sich nahtlos in dieselbe Kette ein.

2. MiddlewareInterface implementieren: Das Grundgerüst

Eine eigene Middleware implementiert MiddlewareInterface mit genau einer Methode, handle(Envelope $envelope, StackInterface $stack): Envelope. Innerhalb dieser Methode entscheidet man, was vor dem Aufruf der nächsten Middleware passiert, ruft dann $stack->next()->handle($envelope, $stack) auf, um die Verarbeitung fortzusetzen, und kann danach noch Code ausführen, der erst nach Abschluss aller nachfolgenden Middleware und des Handlers läuft. Wird $stack->next()->handle() nicht aufgerufen, bricht die gesamte Verarbeitungskette ab, was für bestimmte Fälle wie Deduplizierung durchaus gewünscht sein kann.

Das folgende Beispiel misst die Ausführungszeit jeder Message und loggt sie strukturiert, inklusive des Message-Klassennamens. Wichtig ist der try/finally-Block: Die Zeitmessung soll auch dann korrekt geloggt werden, wenn der Handler eine Exception wirft, denn gerade fehlgeschlagene Verarbeitungen sind für die Performance-Analyse oft die interessantesten Fälle.

<?php

declare(strict_types=1);

namespace App\Messenger\Middleware;

use Psr\Log\LoggerInterface;
use Symfony\Component\Messenger\Envelope;
use Symfony\Component\Messenger\Middleware\MiddlewareInterface;
use Symfony\Component\Messenger\Middleware\StackInterface;

final class ExecutionTimeMiddleware implements MiddlewareInterface
{
    public function __construct(
        private readonly LoggerInterface $logger,
    ) {
    }

    public function handle(Envelope $envelope, StackInterface $stack): Envelope
    {
        $messageClass = $envelope->getMessage()::class;
        $startedAt = microtime(true);

        try {
            return $stack->next()->handle($envelope, $stack);
        } finally {
            $durationMs = (int) round((microtime(true) - $startedAt) * 1000);

            $this->logger->info('Message verarbeitet', [
                'message' => $messageClass,
                'duration_ms' => $durationMs,
            ]);
        }
    }
}

3. Reihenfolge der Middleware-Kette im Messenger konfigurieren

Die Middleware-Kette wird pro Bus in der messenger.yaml unter dem Schlüssel middleware als geordnete Liste konfiguriert, wobei die Reihenfolge in der Konfiguration exakt der Ausführungsreihenfolge entspricht. Eine Middleware, die früh in der Liste steht, läuft entsprechend früh vor dem Handler und spät danach, während eine Middleware am Ende der Liste am nächsten am eigentlichen Handler-Aufruf sitzt. Diese Reihenfolge ist keine Formalität, sondern hat direkte fachliche Konsequenzen.

Eine Logging-Middleware sollte typischerweise sehr früh in der Kette stehen, damit sie auch Fehler erfasst, die von späteren Middleware-Komponenten ausgelöst werden. Eine Transaktions-Middleware dagegen sollte möglichst spät stehen, kurz vor dem Handler, damit die Transaktion nur den eigentlichen Geschäftslogik-Aufruf umschließt und nicht zusätzlich Zeit für Logging oder Metriken innerhalb der offenen Transaktion verbraucht wird. Symfony erlaubt es, eigene Middleware über App\Messenger\Middleware\ExecutionTimeMiddleware als vollqualifizierten Klassennamen direkt zwischen die eingebauten Middleware zu platzieren.

4. Middleware vs. Event-Listener: Wann was

Der Messenger-Bus feuert zusätzlich zur Middleware-Kette auch Events wie WorkerMessageReceivedEvent oder WorkerMessageHandledEvent, was auf den ersten Blick wie eine Überschneidung wirkt. Der wesentliche Unterschied liegt in der Kontrolle über den Ausführungsfluss: Middleware kann die Verarbeitung aktiv anhalten, verändern oder komplett abbrechen, indem sie den Aufruf des nächsten Elements im Stack schlicht auslässt, während ein Event-Listener nur reagiert, nachdem etwas bereits passiert ist, ohne den eigentlichen Ablauf direkt zu steuern.

Middleware ist daher die richtige Wahl für Belange, die den Verarbeitungsfluss selbst beeinflussen sollen, etwa eine Transaktion, die bei einem Fehler ein Rollback auslöst, oder eine Deduplizierung, die eine bereits verarbeitete Message gar nicht erst an den Handler weiterreicht. Event-Listener eignen sich besser für rein beobachtende Zwecke, die unabhängig vom Erfolg oder Misserfolg der Verarbeitung sind, etwa das Aktualisieren eines Dashboards mit der Anzahl verarbeiteter Messages pro Minute, ohne dass dabei in den eigentlichen Verarbeitungsfluss eingegriffen wird.

5. Praxisbeispiel: Ausführungszeit-Messung als Grundlage für Metriken

Die weiter oben gezeigte ExecutionTimeMiddleware lässt sich leicht erweitern, um die gemessene Dauer nicht nur zu loggen, sondern an ein Metriksystem wie Prometheus oder StatsD weiterzugeben, indem im finally-Block zusätzlich ein Histogramm mit dem Message-Klassennamen als Label befüllt wird. Damit lassen sich in einem Dashboard direkt Fragen beantworten wie 'Welcher Message-Typ dauert im 95. Perzentil am längsten' oder 'Ist die durchschnittliche Verarbeitungszeit nach dem letzten Deployment gestiegen', ganz ohne dass ein einziger Handler jemals von Metrik-Code wissen muss.

Für asynchron verarbeitete Messages, die über einen Transport wie Doctrine oder AMQP laufen, ist zusätzlich relevant, ob die Zeitmessung die Wartezeit in der Queue mit einschließen soll oder nur die reine Verarbeitungsdauer beim Worker. Da die Middleware-Kette sowohl beim Dispatch als auch beim tatsächlichen Empfangen der Message durch den Worker durchlaufen wird, empfiehlt es sich, über einen eigenen Envelope-Stamp den Dispatch-Zeitpunkt zu speichern und die Wartezeit separat von der Verarbeitungszeit auszuweisen.

6. Middleware für Transaktions-Wrapping

Ein häufiger Anwendungsfall für eigene Middleware ist das Umschließen der Handler-Ausführung mit einer Datenbank-Transaktion, sodass entweder alle innerhalb des Handlers vorgenommenen Änderungen persistiert werden oder, im Fehlerfall, keine davon. Symfony liefert dafür bereits DoctrineTransactionMiddleware mit, das genau dieses Verhalten für den Standardfall abdeckt, doch bei komplexeren Szenarien mit mehreren Entity-Managern oder zusätzlicher Logik rund um Commit und Rollback lohnt sich eine eigene Variante.

Wichtig bei einer eigenen Transaktions-Middleware ist, die Transaktion wirklich erst unmittelbar vor $stack->next()->handle() zu öffnen und im finally-Block oder explizit im catch-Zweig zu schließen, damit sie so kurz wie möglich offen bleibt und nicht versehentlich auch Code aus vorgelagerten Middleware-Komponenten mit umschließt, die davon nichts wissen. Eine zu lang offene Transaktion erhöht das Risiko von Deadlocks bei parallel laufenden Workern erheblich, besonders bei hoher Nachrichtenfrequenz.

7. Middleware für strukturiertes Logging

Neben reiner Zeitmessung eignet sich Middleware hervorragend, um jede verarbeitete Message strukturiert zu loggen, inklusive relevanter Metadaten wie der Message-ID, dem Transport-Namen aus einem TransportMessageIdStamp, und im Fehlerfall der vollständigen Exception-Kette. Diese zentrale Stelle stellt sicher, dass jede Message unabhängig vom jeweiligen Handler-Autor konsistent geloggt wird, statt dass jeder Handler sein eigenes, leicht abweichendes Logging-Format mitbringt.

Ein häufiger Fehler bei Logging-Middleware ist, sensible Daten wie Passwörter oder Zahlungsinformationen ungefiltert mitzuloggen, nur weil sie Teil des Message-Objekts sind. Es lohnt sich, eine explizite Allowlist an Feldern zu definieren, die in den Log-Kontext aufgenommen werden dürfen, oder Messages, die sensible Daten tragen, ein eigenes Interface implementieren zu lassen, das eine toLogContext()-Methode vorschreibt und so die Kontrolle über das Logging beim jeweiligen Message-Autor belässt.

8. Envelope-Stamps als Kommunikationskanal zwischen Middleware

Da mehrere Middleware-Komponenten nacheinander an demselben Envelope arbeiten, brauchen sie einen Weg, Informationen untereinander auszutauschen, ohne globale Variablen oder Request-Attribute zu missbrauchen. Genau dafür existieren Stamps: kleine, unveränderliche Wertobjekte, die per $envelope->with(new MeinStamp(...)) an den Envelope angehängt werden und von jeder nachfolgenden Middleware per $envelope->last(MeinStamp::class) wieder ausgelesen werden können.

Ein praktisches Beispiel ist ein eigener DispatchedAtStamp, der beim ersten Dispatch den aktuellen Zeitpunkt speichert, sodass eine spätere Middleware beim tatsächlichen Verarbeiten die reine Wartezeit in der Queue berechnen kann, indem sie die Differenz zum aktuellen Zeitpunkt bildet. Da Envelopes unveränderlich sind, gibt with() stets eine neue Instanz zurück, die man konsequent weiterreichen muss, ein häufiger Anfängerfehler ist, das Ergebnis von with() zu verwerfen und mit dem alten Envelope weiterzuarbeiten.

9. Middleware testen ohne echten Transport

Eine Middleware lässt sich isoliert testen, indem man sie direkt instanziiert und mit einem selbst gebauten Envelope sowie einem Test-Double für StackInterface aufruft, das entweder einen einfachen Erfolgs-Handler oder eine Exception simuliert. So kann man in einem PHPUnit-Test prüfen, ob die Middleware bei einer erfolgreichen Verarbeitung korrekt loggt, ob sie im Fehlerfall die Exception unverändert weiterreicht, statt sie zu verschlucken, und ob sie den Envelope korrekt mit den erwarteten Stamps zurückgibt.

Für Integrationstests, die die gesamte Kette inklusive der eigenen Middleware prüfen sollen, bietet sich der InMemoryTransport an, der Messages nicht wirklich über einen externen Broker verschickt, sondern lokal im Speicher hält. Damit lässt sich der komplette Dispatch-Vorgang inklusive aller registrierten Middleware end-to-end durchspielen, ohne dass für den Test eine echte Message-Queue-Infrastruktur wie RabbitMQ oder Redis laufen muss.

10. Zusammenfassung

11. FAQ: Symfony Messenger-Middleware: Das Wichtigste auf einen Blick