RabbitMQ i Symfony: Routing i Exchange Types w praktyce

RabbitMQ jest jednym z najczęściej wykorzystywanych systemów kolejkowania wiadomości, szczególnie popularnym w połączeniu z frameworkiem Symfony. W dzisiejszym poście zajmiemy się głębiej tematyką exchange types w RabbitMQ i ich zastosowaniem z Symfony Messenger. Omówimy różne typy exchange – Direct, Fanout, Topic oraz Headers – oraz pokażemy jak te typy działają w praktyce.

Co to jest Exchange?

Exchange w RabbitMQ jest komponentem odpowiedzialnym za odbieranie wiadomości i kierowanie ich do odpowiednich kolejek na podstawie tzw. routing keys. Exchange nie przechowuje wiadomości, lecz decyduje, do której kolejki mają one trafić. Aby to zrobić, wykorzystuje różne typy exchange, które określają sposób kierowania wiadomości do kolejek.

W RabbitMQ mamy cztery podstawowe typy exchange:

Direct Exchange
Fanout Exchange
Topic Exchange
Headers Exchange

Każdy z tych typów ma swoje specyficzne zastosowanie i logikę działania. Zobaczmy, jak można ich użyć w praktyce w połączeniu z Symfony Messenger.

Konfiguracja RabbitMQ w Symfony Messenger

Symfony Messenger to potężne narzędzie do implementacji asynchronicznych zadań i obsługi wiadomości. Symfony Messenger działa w oparciu o transportery (transports), a jednym z nich może być RabbitMQ.

Konfiguracja transportu RabbitMQ jest bardzo prosta. Możemy skonfigurować ją w pliku messenger.yaml lub config/packages/messenger.yaml:

				
					
framework:
    messenger:
        transports:
            async:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                options:
                    exchange:
                        name: 'my_exchange'
                        type: 'direct'

Tutaj używamy zmiennej środowiskowej MESSENGER_TRANSPORT_DSN, która może wyglądać jak:

				
					MESSENGER_TRANSPORT_DSN=amqp://guest:guest@localhost:5672/%2f

Teraz przejdźmy do omówienia typów exchange.

Direct Exchange

Direct Exchange to najprostszy rodzaj exchange, który kieruje wiadomości do konkretnych kolejek na podstawie routing key.

W Symfony Messenger możesz określić nazwę routing key podczas wysyłania wiadomości:

				
					use Symfony\Component\Messenger\MessageBusInterface;

class NotificationSender
{
    private $bus;

    public function __construct(MessageBusInterface $bus)
    {
        $this->bus = $bus;
    }

    public function sendNotification($messageContent)
    {
        $message = new NotificationMessage($messageContent);
        $this->bus->dispatch($message, [
            'routing_key' => 'user.notifications',
        ]);
    }
}

W przypadku Direct Exchange wiadomość trafia tylko do tych kolejek, które są przypisane do danej routing key. Przykład ten nadaje wiadomość z kluczem user.notifications, co sprawia, że trafi ona tylko do tej kolejki, która jest powiązana z tym kluczem.

W przypadku bardziej złożonych systemów, możemy stworzyć kilka kolejek, np.:

user.notifications
admin.notifications

Każda z tych kolejek może odbierać wiadomości na podstawie swojego własnego klucza.

Fanout Exchange

Fanout Exchange jest używany wtedy, gdy chcesz wysłać wiadomość do wszystkich kolejek. Ten typ exchange ignoruje routing key i po prostu przekazuje wiadomość do wszystkich powiązanych kolejek.

Przykładowa konfiguracja w Symfony Messenger:

				
					
framework:
    messenger:
        transports:
            async:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                options:
                    exchange:
                        name: 'my_fanout_exchange'
                        type: 'fanout'

Przykład użycia w Symfony:

				
					use Symfony\Component\Messenger\MessageBusInterface;

class BroadcastSender
{
    private $bus;

    public function __construct(MessageBusInterface $bus)
    {
        $this->bus = $bus;
    }

    public function broadcastMessage($content)
    {
        $message = new BroadcastMessage($content);
        $this->bus->dispatch($message);
    }
}

W powyższym przykładzie wiadomość wysłana za pomocą Fanout Exchange zostanie przekazana do wszystkich kolejek powiązanych z tym exchange, niezależnie od wartości routing key. Jest to świetne rozwiązanie do szerokiego przesyłania wiadomości, np. do logowania lub powiadamiania użytkowników.

Topic Exchange

Topic Exchange jest najbardziej elastycznym typem exchange, który umożliwia kierowanie wiadomości na podstawie wzorców w kluczach routingu. Routing key może zawierać symbole wieloznaczne, takie jak * (dla jednej części) i # (dla wielu części).

Konfiguracja w Symfony Messenger:

				
					
framework:
    messenger:
        transports:
            async:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                options:
                    exchange:
                        name: 'my_topic_exchange'
                        type: 'topic'

Przykład użycia:

				
					use Symfony\Component\Messenger\MessageBusInterface;

class OrderProcessor
{
    private $bus;

    public function __construct(MessageBusInterface $bus)
    {
        $this->bus = $bus;
    }

    public function processOrder($orderType)
    {
        $message = new OrderMessage($orderType);
        $routingKey = 'order.' . $orderType;
        $this->bus->dispatch($message, [
            'routing_key' => $routingKey,
        ]);
    }
}

Załóżmy, że mamy kolejki związane z różnymi typami zamówień:

Kolejka order.electronics odbiera tylko wiadomości związane z elektroniką.
Kolejka order.* odbiera wszystkie wiadomości związane z zamówieniami jakiegokolwiek rodzaju.

Dzięki Topic Exchange możemy precyzyjnie kontrolować, jakie kolejki otrzymają daną wiadomość.

Headers Exchange

Headers Exchange różni się od innych rodzajów exchange, ponieważ nie opiera się na routing key, ale na nagłówkach wiadomości. Pozwala to na kierowanie wiadomości w zależności od zestawu wartości w nagłówkach, co daje dużą elastyczność.

W Symfony Messenger konfiguracja tego typu wygląda następująco:

				
					
framework:
    messenger:
        transports:
            async:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                options:
                    exchange:
                        name: 'my_headers_exchange'
                        type: 'headers'

Wysyłanie wiadomości z nagłówkami:

				
					use Symfony\Component\Messenger\MessageBusInterface;
use Symfony\Component\Messenger\Stamp\TransportConfigurationStamp;

class HeaderBasedSender
{
    private $bus;

    public function __construct(MessageBusInterface $bus)
    {
        $this->bus = $bus;
    }

    public function sendWithHeaders($content, array $headers)
    {
        $message = new HeaderMessage($content);
        $stamp = new TransportConfigurationStamp(['headers' => $headers]);
        $this->bus->dispatch($message, [$stamp]);
    }
}

Możemy zdefiniować warunki dla nagłówków, np.:

Kolejka high_priority odbiera wiadomości tylko wtedy, gdy nagłówek priority ma wartość high.
Kolejka department_sales odbiera wiadomości z nagłówkiem department: sales.

Dzięki temu mamy bardzo precyzyjną kontrolę nad tym, które wiadomości trafiają do jakich kolejek.

Podsumowanie

RabbitMQ oferuje wiele rodzajów exchange, z których każdy ma swoje specyficzne zastosowanie:

Direct Exchange — dokładne dopasowanie routing key do kolejki.
Fanout Exchange — broadcast do wszystkich kolejek, idealne do powiadomień.
Topic Exchange — elastyczność dzięki routing key z symbolami wieloznacznymi, pozwala na tworzenie zaawansowanych wzorców.
Headers Exchange — oparty na nagłówkach, daje bardzo precyzyjną kontrolę nad kierowaniem wiadomości.

W połączeniu z Symfony Messenger, typy exchange mogą znacząco zwiększyć elastyczność Twojej aplikacji, umożliwiając precyzyjne kierowanie wiadomości i skalowalność komunikacji. Mam nadzieję, że ten post pomógł Ci zrozumieć, jak efektywnie wykorzystać RabbitMQ w połączeniu z Symfony!