Введение в FastStream и асинхронную коммуникацию

Почему современные микросервисы не могут жить без очередей сообщений и как FastStream делает работу с ними такой же простой, как создание FastAPI-приложений.

#Мотивация

Представьте типичную микросервисную архитектуру:

┌─────────────┐      HTTP      ┌─────────────┐
│   Frontend  │ ────────────>  │ API Gateway │
└─────────────┘                └─────────────┘
                                      │
                    ┌─────────────────┼─────────────────┐
                    │                 │                 │
                    ▼                 ▼                 ▼
            ┌──────────────┐  ┌──────────────┐  ┌───────────────┐
            │  User Service│  │ Order Service│  │Payment Service│
            └──────────────┘  └──────────────┘  └───────────────┘

Проблема: сервисы вызывают друг друга синхронно по HTTP. Что происходит, если Payment Service упал?

Order Service ждёт таймаут (30 секунд?)
Пользователь видит «висящий» запрос
Цепочка вызовов падает каскадом

Решение: асинхронная коммуникация через брокер сообщений:

┌──────────────┐         ┌─────────────┐         ┌───────────────┐
│ Order Service│ ──────> │   Broker    │ ──────> │Payment Service│
│  (publisher) │  queue  │  (RabbitMQ) │  queue  │  (subscriber) │
└──────────────┘         └─────────────┘         └───────────────┘

Теперь:

Order Service публикует событие order.created и сразу отвечает пользователю
Payment Service получает сообщение, когда готов
Если Payment Service упал — сообщения накапливаются в очереди и обрабатываются после восстановления

#Что такое брокер сообщений?

Брокер сообщений (Message Broker) — промежуточное ПО, которое управляет передачей сообщений между приложениями.

Publisher ──> [Broker] ──> Subscriber

Задачи брокера:

Буферизация: сообщения хранятся в очереди, пока потребитель не обработает
Гарантия доставки: сообщения не теряются при падении потребителя
Масштабирование: несколько потребителей обрабатывают одну очередь параллельно
Развязка: издатель не знает о подписчике (нет прямой зависимости)

#Основные паттерны коммуникации

#1. Point-to-Point (Очередь задач)

Главное правило: сообщение получает ровно один потребитель.

              ┌─────────────┐
Publisher ──> │   Queue     │ ──> Consumer A ✅ (получил)
              │  (orders)   │ ──> Consumer B ❌ (ждёт следующее)
              └─────────────┘

Как работает:

Несколько потребителей слушают одну и ту же очередь
Брокер отдаёт каждое сообщение только одному потребителю (round-robin)
Если потребитель упал — сообщение вернётся в очередь и достанется другому

Сценарий: распределение нагрузки между воркерами. Например, 1000 заказов и 5 воркеров — каждый воркер берёт заказы из общей кучи, ни один заказ не обработается дважды.

Пример:

# Publisher: отправка заказа в очередь
await broker.publish({"order_id": 123}, "orders")

# Consumer A ИЛИ Consumer B получит сообщение (только один!)
@broker.subscriber("orders")
async def process_order(order: dict):
    print(f"Processing order {order['order_id']}")
    # Этот код выполнится ровно один раз для каждого заказа

Аналогия из жизни: очередь в банке — один клиент идёт к следующему свободному оператору. Два оператора не обслуживают одного клиента одновременно.

Главное правило: сообщение получают все подписчики.

              ┌───────────────────────┐
Publisher ──> │     Topic/Exchange    │ ──> Subscriber A (email) ✅
              │   (order.created)     │ ──> Subscriber B (analytics) ✅
              └───────────────────────┘ └─> Subscriber C (inventory) ✅

Как работает:

Каждый подписчик создаёт свою собственную очередь, привязанную к топику
Брокер копирует сообщение во все привязанные очереди
Каждый подписчик обрабатывает сообщение независимо

Сценарий: событие «заказ создан» интересует разные сервисы — email-сервис хочет отправить подтверждение, аналитика хочет посчитать метрики, склад хочет зарезервировать товар. Все три сервиса должны получить событие.

Пример:

# Publisher: событие "заказ создан"
await broker.publish({"order_id": 123}, "order.created")

# ВСЕ подписчики получат копию сообщения
@broker.subscriber("order.created", "email-service")
async def send_email(event: dict):
    print("Отправляю email...")

@broker.subscriber("order.created", "analytics-service")
async def track_event(event: dict):
    print("Записываю в аналитику...")

@broker.subscriber("order.created", "inventory-service")
async def reserve_item(event: dict):
    print("Резервирую товар на складе...")

Аналогия из жизни: YouTube-канал — все подписчики получают уведомление о новом видео. Одно видео — миллионы просмотров, каждый смотрит сам.

#Сравнение в таблице

Критерий	Point-to-Point	Publish-Subscribe
Кто получает?	Один потребитель	Все подписчики
Очереди	Одна общая	У каждого своя
Сообщений на 1 публикацию	1 обработка	N обработок (по числу подписчиков)
Цель	Распараллеливание нагрузки	Уведомление всех заинтересованных систем
Что если потребителей 2?	Каждый получит ~50% сообщений	Каждый получит 100% сообщений
Пример из жизни	Очередь задач	Подписка на рассылку

#3. Request-Reply (RPC)

Синхронный вызов через очередь.

Client ──> [Request Queue] ──> Server
  ▲                              │
  │         [Reply Queue] <──────┘
  └──────────────────────────────┘

Сценарий: удалённые вычисления, запрос данных у другого сервиса.

Пример:

# RPC-вызов
result = await broker.request(
    "calculate.total",
    {"items": [...]},
    timeout=5.0
)

#Почему FastStream?

#Проблема: работа с брокерами сложная

# «Сырой» RabbitMQ с aio-pika
import aio_pika

async def setup_rabbitmq():
    connection = await aio_pika.connect_robust("amqp://localhost")
    channel = await connection.channel()
    
    queue = await channel.declare_queue("orders", durable=True)
    exchange = await channel.declare_exchange("orders.x", aio_pika.ExchangeType.DIRECT)
    
    await queue.bind(exchange, routing_key="new")
    
    async with queue.iterator() as queue_iter:
        async for message in queue_iter:
            async with message.process():
                await handle(message.body)

Много бойлерплейта, нет типизации, сложно тестировать.

#Решение: FastStream

from faststream import FastStream
from faststream.rabbit import RabbitBroker
from pydantic import BaseModel

class Order(BaseModel):
    id: int
    amount: float

broker = RabbitBroker("amqp://localhost")
app = FastStream(broker)

@broker.subscriber("orders")
async def handle_order(order: Order):  # Автоматическая валидация!
    print(f"Order {order.id}: ${order.amount}")
    await broker.publish({"status": "processed"}, "orders.processed")

Преимущества FastStream:

Единый API для RabbitMQ, Kafka, Redis, NATS
Типизация: Pydantic-модели для входящих сообщений
Dependency Injection: как в FastAPI
Автодокументирование: генерация AsyncAPI-спецификации
Тестирование: TestClient без реального брокера
Middleware: логирование, метрики, трассировка

#Архитектура FastStream

┌─────────────────────────────────────────────────────┐
│                  FastStream App                     │
├─────────────────────────────────────────────────────┤
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐  │
│  │   Rabbit    │  │    Kafka    │  │    Redis    │  │
│  │   Broker    │  │   Broker    │  │   Broker    │  │
│  └─────────────┘  └─────────────┘  └─────────────┘  │
│                        │                            │
│              ┌─────────┴─────────┐                  │
│              │   Core Engine     │                  │
│              │  (subscribers,    │                  │
│              │   publishers,     │                  │
│              │   validation)     │                  │
│              └───────────────────┘                  │
└─────────────────────────────────────────────────────┘

Ключевые компоненты:

Broker — адаптер к конкретному брокеру (RabbitBroker, KafkaBroker, etc.)
@subscriber — декоратор для регистрации обработчиков
@publisher — декоратор для объявления издателей
Pydantic-валидация — автоматический парсинг JSON в модели
Dependency Injection — внедрение зависимостей через Depends

#Сравнение брокеров

Критерий	RabbitMQ	Kafka	Redis Pub/Sub	NATS
Модель	Очереди + Exchanges	Лог событий (topics)	Каналы	Subjects
Персистентность	Да (очереди)	Да (лог)	Нет	Опционально (JetStream)
Throughput	~10K msg/s	~1M msg/s	~50K msg/s	~500K msg/s
Latency	~1ms	~10ms	<1ms	<1ms
Сложная маршрутизация	Да (exchanges)	Нет	Нет	Wildcards
Replay событий	Нет	Да	Нет	С JetStream
Идеально для	Task queues, RPC	Event sourcing, стримы	Real-time уведомления	High-performance сервисы

#Когда что использовать

RabbitMQ:

Сложная маршрутизация (routing keys, headers)
Гарантированная доставка с ACK/NACK
RPC между сервисами
Dead letter queues для обработки ошибок

Kafka:

Поток событий (event stream)
Хранение истории (retention до лет)
Multiple consumers одного топика (CQRS)
Event sourcing и audit log

Redis Pub/Sub:

Простые real-time уведомления
Чаты, онлайн-статусы
Временные данные (кэш + pub/sub)
Когда потеря сообщений допустима

NATS:

Максимальная производительность
Простая модель subjects
Service mesh, service discovery
JetStream для персистентности

┌─────────────┐ HTTP ┌─────────────┐ │ Frontend │ ────────────> │ API Gateway │ └─────────────┘ └─────────────┘ │ ┌─────────────────┼─────────────────┐ │ │ │ ▼ ▼ ▼ ┌──────────────┐ ┌──────────────┐ ┌───────────────┐ │ User Service│ │ Order Service│ │Payment Service│ └──────────────┘ └──────────────┘ └───────────────┘

┌──────────────┐ ┌─────────────┐ ┌───────────────┐ │ Order Service│ ──────> │ Broker │ ──────> │Payment Service│ │ (publisher) │ queue │ (RabbitMQ) │ queue │ (subscriber) │ └──────────────┘ └─────────────┘ └───────────────┘

┌─────────────┐ Publisher ──> │ Queue │ ──> Consumer A ✅ (получил) │ (orders) │ ──> Consumer B ❌ (ждёт следующее) └─────────────┘

┌───────────────────────┐ Publisher ──> │ Topic/Exchange │ ──> Subscriber A (email) ✅ │ (order.created) │ ──> Subscriber B (analytics) ✅ └───────────────────────┘ └─> Subscriber C (inventory) ✅

Критерий

Point-to-Point

Publish-Subscribe

Кто получает?

Один потребитель

Все подписчики

Очереди

Одна общая

У каждого своя

Сообщений на 1 публикацию

1 обработка

N обработок (по числу подписчиков)

Цель

Распараллеливание нагрузки

Уведомление всех заинтересованных систем

Что если потребителей 2?

Каждый получит ~50% сообщений

Каждый получит 100% сообщений

Пример из жизни

Очередь задач

Подписка на рассылку

Client ──> [Request Queue] ──> Server ▲ │ │ [Reply Queue] <──────┘ └──────────────────────────────┘

┌─────────────────────────────────────────────────────┐ │ FastStream App │ ├─────────────────────────────────────────────────────┤ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │ │ Rabbit │ │ Kafka │ │ Redis │ │ │ │ Broker │ │ Broker │ │ Broker │ │ │ └─────────────┘ └─────────────┘ └─────────────┘ │ │ │ │ │ ┌─────────┴─────────┐ │ │ │ Core Engine │ │ │ │ (subscribers, │ │ │ │ publishers, │ │ │ │ validation) │ │ │ └───────────────────┘ │ └─────────────────────────────────────────────────────┘

Критерий

RabbitMQ

Kafka

Redis Pub/Sub

NATS

Модель

Очереди + Exchanges

Лог событий (topics)

Каналы

Subjects

Персистентность

Да (очереди)

Да (лог)

Нет

Опционально (JetStream)

Throughput

~10K msg/s

~1M msg/s

~50K msg/s

~500K msg/s

Latency

~1ms

~10ms

<1ms

Сложная маршрутизация

Да (exchanges)

Нет

Wildcards

Replay событий

Нет

Да

Нет

С JetStream

Идеально для

Task queues, RPC

Event sourcing, стримы

Real-time уведомления

High-performance сервисы

Введение в FastStream и асинхронную коммуникацию