Что такое Exchange в RabbitMQ?

# Exchange в RabbitMQ

Exchange (точка обмена) — это компонент RabbitMQ, который принимает сообщения от производителей (publishers) и маршрутизирует их в соответствующие очереди (queues) на основе правил маршрутизации (binding).

Основная концепция

RabbitMQ работает по следующей схеме:

Publisher → Exchange → Binding rules → Queues → Consumers

Exchange — это промежуточный слой, который:

• Получает сообщения
• Анализирует их
• Решает, в какую очередь отправить
• Отправляет в нужные очереди

Типы Exchange

В RabbitMQ есть 4 основных типа обмена:

1 Direct Exchange

Маршрутизирует сообщения на основе точного совпадения routing key.

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# Создаем Direct Exchange
channel.exchange_declare(exchange='orders_exchange', exchange_type='direct')

# Создаем очередь
channel.queue_declare(queue='orders_queue')

# Привязываем очередь к exchange с routing key
channel.queue_bind(exchange='orders_exchange', queue='orders_queue', routing_key='order.created')

# Публикуем сообщение
channel.basic_publish(
    exchange='orders_exchange',
    routing_key='order.created',
    body='New order data'
)

Когда использовать: когда нужно отправить сообщение в конкретный сервис.

2 Fanout Exchange

Рассылает сообщение во все привязанные очереди, игнорируя routing key.

# Создаем Fanout Exchange
channel.exchange_declare(exchange='notifications', exchange_type='fanout')

# Привязываем несколько очередей
channel.queue_declare(queue='email_notifications')
channel.queue_declare(queue='sms_notifications')
channel.queue_declare(queue='push_notifications')

channel.queue_bind(exchange='notifications', queue='email_notifications')
channel.queue_bind(exchange='notifications', queue='sms_notifications')
channel.queue_bind(exchange='notifications', queue='push_notifications')

# Публикуем — сообщение придет во все три очереди
channel.basic_publish(
    exchange='notifications',
    routing_key='',  # Игнорируется
    body='Important notification for everyone'
)

Когда использовать: broadcast сообщений (уведомления, события).

3 Topic Exchange

Маршрутизирует на основе pattern matching routing key.

# Создаем Topic Exchange
channel.exchange_declare(exchange='logs', exchange_type='topic')

# Очереди подписываются на паттерны
channel.queue_declare(queue='error_logs')
channel.queue_declare(queue='all_logs')

channel.queue_bind(exchange='logs', queue='error_logs', routing_key='error.*')
channel.queue_bind(exchange='logs', queue='all_logs', routing_key='*.*')

# Публикуем сообщения
channel.basic_publish(
    exchange='logs',
    routing_key='error.database',
    body='Database error occurred'
)  # Попадет в error_logs и all_logs

channel.basic_publish(
    exchange='logs',
    routing_key='info.api',
    body='API info message'
)  # Попадет только в all_logs

Паттерны:

• * — одно слово (например, error.database НО НЕ error.db.critical)
• # — ноль или больше слов (например, error.# ловит error, error.db, error.db.critical)

Когда использовать: логирование, аналитика, когда нужна гибкая фильтрация.

4 Headers Exchange

Маршрутизирует на основе заголовков сообщения, а не routing key.

channel.exchange_declare(exchange='orders', exchange_type='headers')

channel.queue_declare(queue='premium_orders')

channel.queue_bind(
    exchange='orders',
    queue='premium_orders',
    arguments={
        'x-match': 'all',  # все условия должны совпадать
        'customer_type': 'premium',
        'priority': 'high'
    }
)

# Публикуем с заголовками
channel.basic_publish(
    exchange='orders',
    routing_key='',
    properties=pika.BasicProperties(
        headers={
            'customer_type': 'premium',
            'priority': 'high'
        }
    ),
    body='Premium order data'
)

Когда использовать: сложные правила маршрутизации по атрибутам.

Сравнение типов

Полный пример с Producer и Consumer

# producer.py
import pika
import json

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.exchange_declare(exchange='user_events', exchange_type='topic')

event_data = {'user_id': 123, 'action': 'registration'}

channel.basic_publish(
    exchange='user_events',
    routing_key='user.registered',
    body=json.dumps(event_data)
)

print("Message sent")
connection.close()

# consumer.py
import pika
import json

def callback(ch, method, properties, body):
    data = json.loads(body)
    print(f"Received: {data}")
    ch.basic_ack(delivery_tag=method.delivery_tag)

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.exchange_declare(exchange='user_events', exchange_type='topic')
channel.queue_declare(queue='user_notifications')
channel.queue_bind(
    exchange='user_events',
    queue='user_notifications',
    routing_key='user.*'
)

channel.basic_consume(
    queue='user_notifications',
    on_message_callback=callback
)

print('Waiting for messages...')
channel.start_consuming()

Ключевые особенности

• Декларирование идемпотентно — можно объявлять одно и то же много раз
• Dead Letter Exchange — для обработки неудачных сообщений
• Persistence — сообщения сохраняются на диск (если настроено)
• Acknowledgments — подтверждение обработки сообщения
• TTL — время жизни сообщения или очереди

Лучшие практики

1. Используй Topic Exchange для микросервисной архитектуры
2. Правильно называй routing keys (например, domain.resource.action)
3. Обрабатывай ошибки и используй retry механизмы
4. Настрой Dead Letter Queue для неудачных сообщений
5. Мониторь очереди и обмены
6. Документируй схему Exchange-Binding-Queue

Итог

Exchange в RabbitMQ — это гибкий механизм маршрутизации, который позволяет создавать различные паттерны обмена сообщениями в распределенных системах. Выбор правильного типа Exchange критичен для архитектуры приложения.