Когда сервисы начинают общаться через брокер, возникает вопрос: как правильно организовать очереди, обменники и подписки? Большинство задач укладываются в несколько типовых схем. Разберём каждую на примерах aio-pika.
Раздать задачи нескольким обработчикам — Work Queue
Представьте: пользователи загружают фотографии, и каждую надо сжать и нарезать в несколько размеров. Это долго. Делать это прямо в HTTP-запросе нельзя — пользователь будет ждать минуты.
Решение — work queue (очередь задач): сохранить задание в очередь и отдать её одному из пула обработчиков.
producer → [одна queue] → consumer 1
→ consumer 2
→ consumer 3
Брокер сам распределяет задачи между обработчиками. Каждое сообщение получает ровно один из них.
channel = await connection.channel()
await channel.set_qos(prefetch_count=20)
images_queue = await channel.declare_queue(
"images.to-process",
durable=True,
arguments={"x-queue-type": "quorum"},
)
async def process(message: AbstractIncomingMessage) -> None:
async with message.process():
job = deserialize(message.body)
# обработка изображения
await images_queue.consume(process)
prefetch_count=20 означает: брокер выдаёт одному потребителю до 20 неподтверждённых сообщений одновременно. Если запустить 5 копий сервиса — получится до 100 параллельных обработчиков.
Когда брать: фоновые задачи — обработка файлов, отправка писем, генерация отчётов, любые «положили в очередь — кто-то возьмёт».
Отправить событие всем сервисам сразу — Publish/Subscribe
Другая задача: произошло событие «конфигурация обновлена» — каждый сервис должен обновить свой кеш. Нельзя знать заранее, кто именно подписан и сколько сервисов запущено.
Это publish/subscribe (рассылка всем): издатель отправляет одно сообщение, а копию получают все подписчики одновременно.
Здесь нужен fanout exchange — он копирует каждое сообщение во все привязанные очереди. Каждый сервис объявляет свою очередь и привязывает её к общему обменнику.
cache_invalidation = await channel.declare_exchange(
"cache.invalidation", ExchangeType.FANOUT, durable=True
)
service_a_cache = await channel.declare_queue(exclusive=True, auto_delete=True)
await service_a_cache.bind(cache_invalidation)
async def invalidate(message: AbstractIncomingMessage) -> None:
async with message.process():
event = deserialize(message.body)
cache.evict(event.key)
await service_a_cache.consume(invalidate)
exclusive=True, auto_delete=True — очередь принадлежит одному соединению и удаляется при отключении. При перезапуске сервиса не накапливается мусор в брокере.
Когда брать: инвалидация кешей, broadcast-уведомления всему кластеру, обновления конфигурации.
Направить событие в нужный обработчик — Routing
Иногда нужно не «всем», а «именно тому, кому надо». Например: событие order.created должно идти в сервис выполнения и в аудит, а order.payment-failed — только в алерты.
Это routing (точечная маршрутизация): direct exchange смотрит на ключ маршрутизации сообщения и отправляет его только в очереди с совпадающим binding key.
orders = await channel.declare_exchange("orders", ExchangeType.DIRECT, durable=True)
quorum = {"x-queue-type": "quorum"}
fulfillment = await channel.declare_queue("orders.fulfillment", durable=True, arguments=quorum)
audit = await channel.declare_queue("orders.audit", durable=True, arguments=quorum)
alerts = await channel.declare_queue("orders.alerts", durable=True, arguments=quorum)
await fulfillment.bind(orders, routing_key="order.created")
await audit.bind(orders, routing_key="order.created")
await audit.bind(orders, routing_key="order.cancelled")
await alerts.bind(orders, routing_key="order.payment-failed")
order.created→ fulfillment + audit.order.cancelled→ только audit.order.payment-failed→ только alerts.
Когда брать: явное разделение потоков — алерты отдельно от аудита, основной обработчик отдельно от мониторинга.
Подписаться по маске — Topic
Routing хорош для жёстких правил. Но что если сервис хочет подписаться на «все события по заказам»? Или «всё из EU-региона»?
Topic exchange позволяет задавать подписки через шаблоны. Ключи сообщений строятся через точку (order.created.eu), а в подписке можно использовать:
*— ровно одно слово,#— ноль и более слов.
events = await channel.declare_exchange("events", ExchangeType.TOPIC, durable=True)
quorum = {"x-queue-type": "quorum"}
audit_all_orders = await channel.declare_queue("audit.orders", durable=True, arguments=quorum)
eu_dashboard = await channel.declare_queue("dashboard.eu", durable=True, arguments=quorum)
alerts = await channel.declare_queue("alerts.critical", durable=True, arguments=quorum)
await audit_all_orders.bind(events, routing_key="order.#")
await eu_dashboard.bind(events, routing_key="*.*.eu")
await alerts.bind(events, routing_key="payment.failed.#")
Сообщение с ключом order.cancelled.eu попадёт в audit_all_orders (по order.#) и в eu_dashboard (по *.*.eu).
Когда брать: события с иерархической структурой, когда нужно гибко подписываться без переделки топологии при добавлении новых типов событий.
Запрос-ответ через очередь — RPC
Иногда нужен синхронный ответ, но HTTP не подходит: сервис за NAT, нет публичного адреса, или хочется балансировки по пулу обработчиков.
RPC через очередь: клиент отправляет запрос и ждёт ответа. Брокер доставляет запрос одному из обработчиков, тот отвечает в отдельную очередь-ответ. Для сопоставления запроса и ответа используется correlation-id.
В aio-pika это скрыто за паттерном RPC:
from aio_pika.patterns import RPC
# Клиент
rpc = await RPC.create(channel)
quote = await rpc.call("pricing.quote", kwargs={"request": request})
# Сервер
async def handle(*, request: QuoteRequest) -> PriceQuote:
return PriceQuote.compute(request) # возврат автоматически уходит в reply-to
rpc = await RPC.create(channel)
await rpc.register("pricing.quote", handle, auto_delete=True)
aio-pika сам создаёт временную очередь-ответ, проставляет reply-to и correlation-id, ждёт ответа. Возврат из зарегистрированного обработчика автоматически публикуется обратно.
Когда брать: нужен синхронный вызов, но HTTP не работает (NAT, firewall, нет публичного адреса); нужна балансировка запросов по пулу обработчиков.
Когда не брать: если HTTP/gRPC просто работает — RPC через брокер сложнее в отладке и дороже.
Что делать с повторной доставкой — Idempotent Consumer
AMQP гарантирует at-least-once доставку: одно и то же сообщение может прийти дважды. Это случается, когда брокер не получил подтверждение обработки (например, из-за сетевой проблемы) и переотправляет сообщение.
Обработчик обязан уметь работать с повторами, не ломая бизнес-логику.
Ключ идемпотентности в базе данных
Самый надёжный способ — запоминать уже обработанные сообщения:
async def process(message: AbstractIncomingMessage) -> None:
async with message.process():
event = deserialize(message.body)
async with session.begin():
if await processed_events_repo.exists(event.idempotency_key):
return # уже обработано — просто подтверждаем получение
await processed_events_repo.save(ProcessedEvent(event.idempotency_key))
await account_repo.debit(event.account_id, event.amount)
Таблица processed_events с уникальным индексом на idempotency_key. Если два одинаковых сообщения придут одновременно — база данных поймает дубль через нарушение уникального ограничения.
Проверка состояния объекта
Если событие переводит объект в новое состояние, достаточно проверить текущее:
async def on_order_confirmed(event: OrderConfirmedEvent) -> None:
async with session.begin():
order = await order_repo.get(event.order_id)
if order.status == OrderStatus.CONFIRMED:
return # уже в нужном состоянии
order.confirm()
await order_repo.save(order)
Не требует отдельной таблицы — состояние уже хранится в бизнес-объекте.
Retry с задержкой и Dead Letter Queue
Что если обработчик упал не из-за бага, а из-за временной недоступности внешнего сервиса? Нужно попробовать снова, но не немедленно.
Delayed retry собирается через x-message-ttl и Dead Letter Exchange:
retry_queue = await channel.declare_queue(
"orders.retry",
durable=True,
arguments={
"x-message-ttl": 30_000, # ждём 30 секунд
"x-dead-letter-exchange": "orders",
"x-dead-letter-routing-key": "order.created",
"x-queue-type": "quorum",
},
)
Поток: обработчик отклонил сообщение → оно попадает в retry очередь → через 30 секунд по истечении TTL уходит через DLX обратно в основную очередь → новая попытка.
Количество попыток считается через заголовок x-death.count — его нужно проверять вручную, встроенного ограничителя нет.
Сообщения, которые не получилось обработать после всех попыток, уходят в Dead Letter Queue (DLQ) — отдельную очередь для разбора вручную или через алерты.
Гарантированная публикация — Outbox
Бывает задача: сохранить заказ в базу данных и опубликовать событие — атомарно. Если сначала сохранить, потом опубликовать, то сервис может упасть между двумя операциями. Событие потеряется.
Outbox pattern: событие сохраняется в ту же транзакцию, что и бизнес-данные. Отдельный процесс читает таблицу и публикует в AMQP.
async def confirm(order_id: OrderId) -> None:
async with session.begin():
order = await order_repo.get(order_id)
order.confirm()
await order_repo.save(order)
await outbox_repo.save(OutboxEvent(
id=uuid.uuid4(),
routing_key="order.confirmed",
exchange="orders",
payload=to_json(OrderConfirmedEvent(order_id)),
))
async def publish_outbox() -> None: # фоновая задача
while True:
async with session.begin():
batch = await outbox_repo.fetch_unpublished(100)
for event in batch:
exchange = await channel.get_exchange(event.exchange)
await exchange.publish(
Message(event.payload.encode()), routing_key=event.routing_key
)
await outbox_repo.mark_published(event.id)
await asyncio.sleep(0.5)
Либо оба изменения зафиксированы, либо ни одного. Дубли возможны (публикация прошла, но пометить как отправленное не успело) — поэтому получатель всё равно должен быть идемпотентным.
Шпаргалка по выбору
| Задача | Паттерн | Тип обменника |
|---|---|---|
| Распределить нагрузку между обработчиками | Work Queue | direct (default) |
| Broadcast события всем сервисам | Publish/Subscribe | fanout |
| Разные события в разные очереди | Routing | direct |
| Подписка по маске на иерархические события | Topic | topic |
| Синхронный вызов через очередь | RPC | direct + reply-to |
| Защита от повторной доставки | Idempotent Consumer | любой |
| Retry с задержкой | Delayed Retry | direct + DLX |
| Атомарная публикация вместе с записью в БД | Outbox | direct |
Коротко
- Work Queue — одна очередь, несколько обработчиков, каждое сообщение получает ровно один. Для фоновых задач.
- Publish/Subscribe — fanout exchange копирует сообщение во все привязанные очереди. Для broadcast-событий.
- Routing — direct exchange смотрит на ключ маршрутизации. Для точного разделения потоков.
- Topic — как routing, но с шаблонами
*и#. Для иерархических событий с гибкой подпиской. - RPC через очередь — запрос-ответ через брокер с
reply-toиcorrelation-id. Для вызовов без HTTP. - Idempotent Consumer — at-least-once означает возможные дубли. Защита: ключ идемпотентности в БД или проверка состояния объекта.
- Delayed Retry — TTL + DLX: сообщение «паркуется» на время, потом возвращается.
- Outbox — событие сохраняется в той же транзакции, что и данные. Атомарность без двухфазного коммита.
Что почитать дальше
- Протокол AMQP — модель exchange/binding/queue изнутри.
- Spring AMQP — конфигурация, RabbitTemplate, аннотации.
- RabbitMQ в production — Quorum Queues, кластеризация, мониторинг.
- AMQP vs Kafka — какой брокер и когда брать.