Mehrere RabbitMQ-Queues in Symfony mit enqueue-bundle einrichten

Wenn Sie in Ihrem Symfony-Projekt Nachrichten über verschiedene Queues in RabbitMQ verarbeiten müssen, bietet die Bibliothek enqueue/enqueue-bundle dafür einen eleganten und flexiblen Weg. Dieser Artikel führt Sie durch die grundlegende Einrichtung mehrerer Queues und deren Verarbeitung mit spezifischen Prozessoren.

1. Grundkonfiguration der Bibliothek enqueue

In Ihrer Konfigurationsdatei (z. B. config/packages/enqueue.yaml) müssen Transport und Client definiert werden. Für die Verbindung zu RabbitMQ verwenden wir einen DSN, der in einer Umgebungsvariablen gespeichert ist. Außerdem definieren wir die Standard-Queue und weitere Client-Parameter:

enqueue:
    default:
        transport: '%env(resolve:ENQUEUE_DSN)%'
        client:
            app_name: "%enqueue_app_name%"
            default_queue: "%enqueue_default_queue%"
            redelivered_delay_time: 5
            router_topic: default
            router_queue: router
            router_processor: null
        job: true

Hinweis: Die Werte enqueue_app_name und enqueue_default_queue legen Sie nach Ihren Bedürfnissen in .env oder .env.local fest.

2. Definition von Prozessoren und Queues

Damit Nachrichten korrekt in die verschiedenen Queues geroutet werden, muss jeder Prozessor die statische Methode getSubscribedTopics implementieren, in der Sie festlegen, welche Topics und Queues der Prozessor bedient. Hier ein Beispiel:

use Enqueue\Client\TopicSubscriberInterface;

class TestProcessor implements TopicSubscriberInterface
{
    public function __invoke(Message $message, Context $context): string
    {
        // Nachricht verarbeiten
        return self::ACK;
    }

    public static function getSubscribedTopics(): array
    {
        return [
            [
                'topic' => 'test.misc',
                'queue' => 'test_queue',
                'prefix_queue' => true,
            ],
        ];
    }
}

Auf diese Weise können Sie festlegen, welche Nachrichten in welche Queue geroutet werden und welcher Prozessor sie verarbeitet.

3. Initialisierung der Queues und Start der Consumer

Nach der Konfiguration müssen die Queues in RabbitMQ angelegt und die Consumer gestartet werden, die die Nachrichten verarbeiten. Gehen Sie wie folgt vor:

a) Anlegen aller benötigten Queues

bin/console enqueue:setup-broker

Mit diesem Befehl werden alle Queues angelegt, einschließlich der Routing-Queue (router), die für die Verteilung der Nachrichten an die richtigen Ziel-Queues zuständig ist.

b) Start des Router-Consumers

Der Router-Consumer lauscht auf der Router-Queue und leitet Nachrichten gemäß den Regeln weiter, die in getSubscribedTopics() der einzelnen Prozessoren definiert sind:

bin/console enqueue:consume router --time-limit="now + 300 seconds"

Tipp: Wir empfehlen, diesen Prozess separat als Daemon zu betreiben, z. B. mit Supervisor oder systemd.

c) Start der Consumer für die einzelnen Queues

Starten Sie für jede einzelne Queue einen eigenen Consumer. Beispiel für die Queue test_queue:

bin/console enqueue:consume test_queue --time-limit="now + 300 seconds"

Dieser Consumer verarbeitet ausschließlich Nachrichten, die in die Queue test_queue geroutet werden.

Zusammenfassung

Dank der Bibliothek enqueue-bundle lassen sich in Symfony ganz einfach mehrere RabbitMQ-Queues einrichten, ihnen spezifische Prozessoren zuordnen und so die Verarbeitung asynchroner Aufgaben effizient skalieren. Dieser Ansatz bewährt sich vor allem bei Projekten mit einem größeren Nachrichtenaufkommen oder unterschiedlichen Arten der Verarbeitung.