Зачем нужен брокер сообщений

Почему сервис, который ходит к соседям напрямую, падает вместе с ними — и что меняет посредник в середине.

Брокер сообщений (message broker) — отдельный сервис-посредник: одни программы кладут в него сообщения, другие забирают их и обрабатывают. Отправитель не знает, кто именно прочитает сообщение и когда, и не ждёт ответа.

Всё начинается с одной невинной строчки

Вы пишете регистрацию пользователя. Сохранили в базу — надо отправить письмо. Логично же: вот прямо тут и отправим. Потом продакт просит записывать регистрацию в аналитику. Потом маркетинг просит создавать лид в CRM. Через три спринта обработчик выглядит так:

# registration.py — наивная версия: всё делаем прямо в обработчике HTTP-запроса
import requests

def register(email, password):
    user = db.create_user(email, password)                                 # ~20 мс
    requests.post("http://mailer/send", json={"to": email})                # ~300 мс
    requests.post("http://analytics/track", json={"event": "signup"})      # ~150 мс
    requests.post("http://crm/lead", json={"email": email})                # ~400 мс
    return user

Код читается прекрасно и работает — ровно до первой пятницы. Разберём, что здесь сломано.

1. Сосед лежит — мы лежим

CRM выкатила релиз и на две минуты ушла в перезагрузку. Что происходит с вашей регистрацией? Она падает. Пользователь уже создан в базе, письмо ушло, событие в аналитику записано — а requests.post в CRM бросает исключение, и клиент видит «500 Internal Server Error». Он жмёт «Зарегистрироваться» ещё раз и получает «email уже занят». Отличный первый опыт.

Заметьте, что ваш сервис при этом полностью исправен. Вы уронили регистрацию из-за чужого сервиса, который к сути регистрации отношения не имеет. Это и называется жёсткой связанностью: доступность вашей функции равна произведению доступности всех, к кому вы ходите. Четыре зависимости по 99,9% дают уже 99,6% — это лишние три часа простоя в год, которые вы себе не заказывали.

2. Медленный сосед делает медленными вас

Сложите тайминги из примера: 20 + 300 + 150 + 400 = 870 мс. Пользователь ждёт почти секунду ради того, что ему не нужно прямо сейчас. Письмо может прийти через 5 секунд — никто не заметит. Но пока код синхронный, воркер вашего веб-сервера занят и не может обслужить следующего клиента. А если SMTP-сервер провайдера задумался на 30 секунд, ваш пул воркеров кончится, и упадёт уже весь сайт — не только регистрация.

3. Пики никто не сглаживает

В «Чёрную пятницу» на вас приходит 5000 заказов в минуту. Сервис уведомлений может переварить 500 в минуту — он упирается в внешний SMS-шлюз, и это не лечится масштабированием. При прямых вызовах лишние 4500 запросов просто получают отказ или таймаут: данные потеряны. Никакой «подожди, я разберу потом» в HTTP не встроено.

Что делает брокер

Брокер вставляется между отправителем и получателем и берёт на себя одну простую обязанность: принять сообщение и не потерять его, пока кто-нибудь не подтвердит обработку. Тот же код с RabbitMQ:

import json
import pika

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

def register(email, password):
    user = db.create_user(email, password)                 # ~20 мс
    channel.basic_publish(                                 # ~1 мс
        exchange="events",
        routing_key="user.registered",
        body=json.dumps({"user_id": user.id, "email": email}),
        properties=pika.BasicProperties(delivery_mode=2),  # переживёт рестарт брокера
    )
    return user

Обработчик стал в сорок раз быстрее и больше ничего не знает про почту, аналитику и CRM. Он объявляет факт: «пользователь зарегистрировался». Кому это интересно — подпишется сам.

Что мы за это получили

Буфер. Сообщения копятся в очереди. CRM лежит две минуты — сообщения ждут её на диске у брокера; поднялась — разгребла отставание. Пользователь ничего не заметил.
Сглаживание пиков. 5000 сообщений в минуту влетают в очередь мгновенно, а воркеры разбирают их со своей скоростью. Очередь растёт и потом рассасывается — это нормальная работа, а не авария.
Слабая связанность. Новому сервису (например, «начислить бонусные баллы») не нужно ничего менять в регистрации: он подписывается на user.registered и живёт своей жизнью. Публикующий код не трогали.
Горизонтальное масштабирование. Очередь копится? Запустите пять воркеров вместо одного — RabbitMQ раздаст им сообщения по очереди, без всякой координации с вашей стороны.

Какие задачи отдают брокеру

Задача	Почему её нельзя делать в HTTP-запросе
Письма и push-уведомления	Внешний SMTP/APNs медленный и регулярно отваливается
Генерация отчётов, PDF, экспорт в Excel	Занимает десятки секунд, HTTP-таймаут не выдержит
Обработка загруженных файлов (превью, транскодинг видео)	Тяжёлый CPU, нельзя занимать веб-воркер
Синхронизация с внешними API (CRM, 1С, платёжка)	Чужая доступность не должна ронять вашу
Отложенные действия («напомнить через сутки»)	Нужен буфер с задержкой, а не удерживаемое соединение

Общий признак: результат нужен не прямо сейчас, и пользователь не должен его ждать. Если же вам нужно вернуть ответ немедленно (проверить пароль, получить корзину), брокер не нужен — это классический синхронный вызов.

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

В RabbitMQ путь сообщения такой: producer (издатель) публикует сообщение не в очередь напрямую, а в exchange (обменник). Exchange по правилам привязок (bindings) решает, в какие очереди положить копию сообщения. Из очереди сообщение забирает consumer (потребитель) и в конце присылает брокеру ack — подтверждение «обработал, можно удалять». Не прислал ack и отвалился — RabbitMQ вернёт сообщение в очередь и отдаст другому воркеру. Именно ack превращает очередь из «трубы» в надёжный механизм: сообщение исчезает только после успешной обработки.

Саму идею буфера легко почувствовать без всякого брокера — вот та же схема на голом Python (этот пример можно запустить прямо здесь):

from collections import deque

queue = deque()

def publish(task):
    queue.append(task)
    print("[api] положил в очередь:", task)

def worker():
    while queue:
        task = queue.popleft()
        print("[worker] обработал:", task)
    print("[worker] очередь пуста")

for name in ["anna", "boris", "vera"]:
    publish("welcome_email:" + name)

print("--- ответ пользователю уже отправлен, ждать никого не пришлось ---")
worker()

Результат:

[api] положил в очередь: welcome_email:anna
[api] положил в очередь: welcome_email:boris
[api] положил в очередь: welcome_email:vera
--- ответ пользователю уже отправлен, ждать никого не пришлось ---
[worker] обработал: welcome_email:anna
[worker] обработал: welcome_email:boris
[worker] обработал: welcome_email:vera
[worker] очередь пуста

Разница с настоящим брокером в том, что deque живёт в памяти одного процесса: перезапустили — потеряли всё. RabbitMQ — это отдельный процесс (обычно на отдельной машине), который умеет писать сообщения на диск, раздавать их конкурирующим воркерам, повторять доставку после сбоя и показывать метрики. Идея та же, гарантии другие.

Частые ошибки

Брокер там, где нужен синхронный ответ. «Проверить промокод» через очередь — это боль: придётся городить очередь ответов и ждать её. Асинхронность хороша для «сделай и забудь», а не для «дай мне число прямо сейчас».
Считать, что брокер «просто ускорил» код. Он не ускорил — он перенёс работу в фон. Письмо всё ещё будет идти 300 мс, просто не за счёт пользователя. Важное следствие: система становится eventually consistent. Сразу после ответа «вы зарегистрированы» письма ещё нет — интерфейс должен это переживать.
Публиковать в очередь ссылку на данные, которых ещё нет. Классика: положили сообщение {"order_id": 42} внутри транзакции, транзакция откатилась, а воркер уже читает заказ 42 и не находит его. Публикуйте после коммита, а лучше — через transactional outbox.
Забыть, что очередь может расти бесконечно. Если воркеры медленнее издателей в среднем, а не только на пике, очередь не рассосётся никогда: она просто съест память и диск брокера. Очередь сглаживает всплески, но не лечит нехватку мощности.
Не думать о повторной доставке. Воркер упал после отправки письма, но до ack — RabbitMQ отдаст сообщение снова, и письмо уйдёт дважды. Обработчики должны быть идемпотентными; к этому мы ещё вернёмся отдельно.

Итоги

Прямые вызовы связывают доступность сервисов: падает сосед — падаете вы, тормозит сосед — тормозите вы.
Брокер — посредник, который принимает сообщение, хранит его и повторяет доставку, пока получатель не подтвердит обработку.
Три главные выгоды: буфер (переживаем простой получателя), сглаживание пиков (нагрузка растягивается во времени), слабая связанность (новых потребителей добавляют без правки издателя).
Типичные кандидаты в очередь: письма, уведомления, отчёты, обработка файлов, синхронизация с внешними API.
Плата за это — eventual consistency, возможные дубликаты и новый инфраструктурный компонент, за которым надо следить.