"The big idea is messaging. [...] The key in making great and growable systems is much more to design how its modules communicate rather than what their internal properties and behaviors should be." — Alan Kay

В предыдущих главах мы рассматривали системы, где компоненты общаются синхронно: клиент отправляет запрос и ждёт ответа. Такая модель проста и предсказуема, но у неё есть фундаментальное ограничение — она создаёт жёсткую связность между компонентами. Если один из сервисов медленный или недоступен, вся цепочка останавливается.

Очереди сообщений и event-driven архитектура решают эту проблему, вводя промежуточный слой — брокер, который принимает сообщения от отправителя и доставляет их получателю асинхронно. Это позволяет развязать компоненты по времени, скорости и доступности.

В этой главе мы разберём фундаментальные концепции асинхронной коммуникации, детально изучим Apache Kafka и RabbitMQ, рассмотрим облачные решения (SQS/SNS), паттерны обмена сообщениями, гарантии доставки и механизмы обработки ошибок.

---

1 Message Queue vs Message Broker vs Event Streaming

Термины «очередь сообщений», «брокер сообщений» и «стриминг событий» часто используются как синонимы, но между ними есть принципиальные различия.

Message Queue (Очередь сообщений)

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

Producer → [Queue: msg1, msg2, msg3] → Consumer
                                        (msg1 удалён после обработки)

Аналогия: очередь в кассу магазина. Каждый покупатель (сообщение) обслуживается одним кассиром (consumer) и после обслуживания покидает очередь.

Message Broker (Брокер сообщений)

Брокер — это программа или сервис, который выступает посредником (translation layer) между отправителями и получателями. Он может маршрутизировать сообщения, трансформировать форматы, фильтровать по правилам и управлять подписками. RabbitMQ — классический пример брокера.

Брокер — это более широкое понятие, чем очередь. Очередь — это один из компонентов внутри брокера.

Event Streaming (Стриминг событий)

Event streaming — это непрерывный поток событий об изменении состояния, обрабатываемый в реальном времени. В отличие от классических очередей, платформы event streaming (Apache Kafka) используют append-only log (журнал): сообщения не удаляются после обработки, а хранятся длительное время, позволяя повторное чтение.

Producer → [Log: e1, e2, e3, e4, e5, ...] ← Consumer A (offset=3)
                                            ← Consumer B (offset=1)
                                            ← Consumer C (offset=5)

Каждый потребитель читает лог независимо, со своим смещением (offset). Потребитель B может перечитать старые события, потребитель C — читать только новые.

Сравнительная таблица

Когда что выбирать:

• Очередь — когда каждая задача должна быть выполнена ровно один раз одним обработчиком (обработка заказов, отправка email, resize изображений).
• Брокер — когда нужна сложная маршрутизация, фильтрация или преобразование сообщений между сервисами.
• Event streaming — когда нужно хранить историю событий, поддерживать множество независимых потребителей и обрабатывать потоки данных в реальном времени (аналитика, CDC, ML-пайплайны).

---

2 Паттерны обмена сообщениями

2.1 Point-to-Point (P2P)

Сообщение доставляется одному и только одному потребителю. После подтверждения обработки сообщение удаляется из очереди.

                    ┌──────────┐
Producer ──msg──▶   │  Queue   │ ──msg──▶ Consumer A
                    └──────────┘   
                            (Consumer B не получит это сообщение)

Использование: Распределение задач между воркерами. Если есть 10 воркеров, каждое сообщение попадёт ровно к одному из них, обеспечивая параллельную обработку без дублирования.

Примеры: обработка заказов, генерация отчётов, отправка уведомлений.

2.2 Publish/Subscribe (Pub/Sub)

Сообщение отправляется в топик и получается всеми подписанными потребителями.

                    ┌──────────┐ ──msg──▶ Subscriber A
Publisher ──msg──▶  │  Topic   │ ──msg──▶ Subscriber B
                    └──────────┘ ──msg──▶ Subscriber C
                    

Использование: Уведомление множества сервисов об одном событии. Например, при создании заказа сообщение «order.created» получают сервис оплаты, сервис доставки, сервис аналитики и сервис уведомлений.

2.3 Request-Reply

Отправитель публикует сообщение и ожидает ответ в отдельной очереди. Реализация синхронного взаимодействия поверх асинхронного транспорта.

Client ──request──▶ [Request Queue] ──▶ Server
Client ◀──reply──── [Reply Queue]   ◀── Server

# Client
correlation_id = uuid4()
channel.publish(
    exchange="",
    routing_key="rpc_queue",
    body=request_data,
    properties=pika.BasicProperties(
        reply_to="amq.rabbitmq.reply-to",
        correlation_id=str(correlation_id),
    )
)
# Ждём ответ с нужным correlation_id

Использование: RPC через очереди, когда нужна асинхронность транспорта, но синхронность логики (например, агрегация данных из нескольких сервисов с таймаутом).

2.4 Гибридная модель Kafka

Kafka элегантно совмещает оба основных паттерна:

• Pub/Sub между consumer groups: Каждая consumer group получает все сообщения из топика
• Point-to-Point внутри consumer group: Внутри группы каждая партиция назначена ровно одному потребителю

Topic: "orders"  (3 partitions)

Consumer Group "payments":
  Consumer A ← Partition 0
  Consumer B ← Partition 1, 2

Consumer Group "analytics":
  Consumer C ← Partition 0, 1, 2

Сообщение в Partition 0:
  → Consumer A (payments) получит
  → Consumer C (analytics) получит
  → Consumer B (payments) НЕ получит (не его партиция)

---

3 Apache Kafka: архитектура и внутреннее устройство

Apache Kafka был создан в LinkedIn для обработки потоков данных (логи, метрики, user activity) и впоследствии передан Apache Software Foundation. Сегодня Kafka — де-факто стандарт для event streaming, обрабатывающий триллионы сообщений в день у таких компаний как LinkedIn, Uber, Netflix и Airbnb.

3.1 Основные компоненты

┌───────────────────────────────────────────────────┐
│                   Kafka Cluster                   │
│                                                   │
│  ┌───────────┐    ┌───────────┐    ┌───────────┐  │
│  │ Broker #1 │    │ Broker #2 │    │ Broker #3 │  │
│  │           │    │           │    │           │  │
│  │ Topic A   │    │ Topic A   │    │ Topic A   │  │
│  │  P0(L)    │    │  P1(L)    │    │  P2(L)    │  │
│  │  P1(F)    │    │  P2(F)    │    │  P0(F)    │  │
│  │           │    │           │    │           │  │
│  │ Topic B   │    │ Topic B   │    │ Topic B   │  │
│  │  P0(F)    │    │  P0(L)    │    │  P1(L)    │  │
│  └───────────┘    └───────────┘    └───────────┘  │
│                                                   │
│  L = Leader, F = Follower                         │
│                                                   │
│    ┌─────────────────────────────────────────┐    │
│    │  ZooKeeper / KRaft (координация)        │    │
│    └─────────────────────────────────────────┘    │
└───────────────────────────────────────────────────┘

Producers ──▶ Brokers ──▶ Consumer Groups

Broker — сервер Kafka, хранящий данные и обслуживающий запросы. Кластер обычно состоит из 3–100+ брокеров.

Topic — именованный канал для категории сообщений. Аналог таблицы в БД. Например: orders, user-events, payment-transactions.

Partition — единица параллелизма и масштабирования. Каждый топик разделён на N партиций, распределённых по брокерам. Внутри партиции сообщения строго упорядочены (FIFO) и идентифицируются смещением (offset).

Partition 0: [msg0, msg1, msg2, msg3, msg4, msg5, ...]
              ↑                              ↑
              offset=0                       offset=5 (последнее)

ZooKeeper / KRaft — компонент координации кластера. Отвечает за выбор лидера партиции, отслеживание здоровья брокеров и хранение метаданных. Начиная с Kafka 3.3+ ZooKeeper заменяется встроенным протоколом KRaft (Kafka Raft), который упрощает развёртывание и устраняет внешнюю зависимость.

3.2 Партиционирование

Партиции — ключ к масштабируемости Kafka. Они позволяют:

• Распараллелить обработку: Каждый consumer в group читает свои партиции
• Распределить данные по брокерам: Партиции размещаются на разных серверах
• Обеспечить порядок: Внутри одной партиции порядок гарантирован

Как сообщение попадает в партицию:

# Стратегии назначения партиции:

# 1. По ключу (Key-based) — сообщения с одинаковым ключом всегда в одной партиции
partition = hash(message.key) % num_partitions
# Пример: ключ = user_id → все события пользователя идут в одну партицию → порядок гарантирован

# 2. Round-Robin — если ключ не указан, сообщения распределяются равномерно
# Порядок между партициями не гарантирован

# 3. Custom Partitioner — кастомная логика
class GeoPartitioner:
    def partition(self, topic, key, value, partitions):
        region = extract_region(value)
        if region == "EU":
            return 0
        elif region == "US":
            return 1
        else:
            return 2

Выбор количества партиций: Больше партиций = больше параллелизма, но и больше overhead (файловые дескрипторы, end-to-end latency при репликации, время восстановления при failover). Практическое правило: начать с количества партиций, равного ожидаемому пиковому числу consumer-ов, с запасом на рост. LinkedIn рекомендует не более нескольких тысяч партиций на один брокер.

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

3.3 Consumer Groups

Consumer group — это механизм, позволяющий группе потребителей совместно обрабатывать топик.

Topic "orders" — 6 партиций

Consumer Group "payment-service" (3 инстанса):
  Consumer A: Partitions 0, 1
  Consumer B: Partitions 2, 3
  Consumer C: Partitions 4, 5

Consumer Group "analytics-service" (2 инстанса):
  Consumer D: Partitions 0, 1, 2
  Consumer E: Partitions 3, 4, 5

Правила назначения:

• Каждая партиция назначается ровно одному consumer-у внутри группы
• Один consumer может обрабатывать несколько партиций
• Если consumer-ов больше, чем партиций — лишние consumer-ы простаивают (idle)
• При падении consumer-а его партиции перераспределяются между оставшимися (rebalancing)

Rebalancing — процесс перераспределения партиций при изменении состава группы (добавление/удаление consumer-а). Во время rebalancing обработка приостанавливается. Kafka поддерживает несколько стратегий:

• Eager (Stop-the-World): все партиции отзываются и перераспределяются заново
• Cooperative/Incremental: только затронутые партиции перераспределяются, остальные продолжают работу. Предпочтительный вариант с Kafka 2.4+.

3.4 Репликация и отказоустойчивость

Kafka использует leader-follower репликацию для обеспечения отказоустойчивости.

Partition 0:
  Broker 1: Leader    ←── Producers пишут сюда
  Broker 2: Follower  ←── Реплицирует с Leader
  Broker 3: Follower  ←── Реплицирует с Leader

  Consumers читают с Leader (по умолчанию)
  или с Follower (KIP-392, Kafka 2.4+, для снижения cross-DC latency)

ISR (In-Sync Replicas) — набор реплик, которые полностью «догнали» лидера. Реплика считается синхронизированной, если она отстаёт от лидера не более чем на заданный порог (параметр replica.lag.max.messages или replica.lag.time.max.ms).

Сообщение считается committed (зафиксированным) только когда оно записано на всех репликах из ISR. Это гарантирует, что при падении лидера новый лидер (выбранный из ISR) будет содержать все зафиксированные сообщения.

Если реплика отстаёт: Она исключается из ISR. Система продолжает работу с меньшим ISR. Когда реплика «догоняет» лидера — она возвращается в ISR. Это позволяет балансировать между производительностью и надёжностью: ожидание всех реплик замедляет работу, но исключение медленных реплик позволяет не блокировать весь раздел.

Конфигурация подтверждений (acks):

from kafka import KafkaProducer

# Максимальная надёжность
producer = KafkaProducer(
    bootstrap_servers=["broker1:9092", "broker2:9092"],
    acks="all",                    # Ждём всех ISR
    retries=5,                     # Повторные попытки
    enable_idempotence=True,       # Идемпотентность (дедупликация)
    max_in_flight_requests_per_connection=5,  # С идемпотентностью можно > 1
)

# Максимальная скорость (fire-and-forget)
producer_fast = KafkaProducer(
    bootstrap_servers=["broker1:9092"],
    acks=0,                        # Не ждём подтверждения
)

Рекомендация: Для критичных данных (платежи, заказы) — acks=all + min.insync.replicas=2. Для логов и метрик — acks=1 или acks=0.

3.5 Хранение данных

В отличие от классических брокеров, Kafka хранит данные на диске. Каждая партиция — это директория с набором segment-файлов.

/kafka-data/topic-orders/partition-0/
  00000000000000000000.log    # Segment: offsets 0-999
  00000000000000000000.index  # Индекс для быстрого поиска по offset
  00000000000000001000.log    # Segment: offsets 1000-1999
  00000000000000001000.index
  ...

Почему Kafka быстр при записи на диск:

1. Sequential I/O: Все записи append-only — данные добавляются в конец файла. Sequential write на HDD — ~600 MB/s, random write — ~100 IOPS. Разница в 3-4 порядка.
2. Page cache ОС: Kafka использует page cache ядра Linux вместо heap JVM. Данные кэшируются в RAM операционной системой.
3. Zero-copy transfer: При отправке данных потребителю Kafka использует sendfile() syscall, который передаёт данные из page cache напрямую в сетевой буфер, минуя копирование в user space.
4. Batching: Producer собирает сообщения в пакеты, снижая overhead на каждое сообщение.

Retention (хранение):

# Хранить данные 7 дней
log.retention.hours=168

# Или ограничить объём
log.retention.bytes=1073741824  # 1 GB на партицию

# Compaction — хранить только последнее значение для каждого ключа
log.cleanup.policy=compact

Log compaction — особый режим, где Kafka хранит только последнее сообщение для каждого ключа. Полезен для changelog-топиков, где важно текущее состояние, а не история (например, user-profiles-changelog).

---

4 RabbitMQ: exchanges, queues, bindings

RabbitMQ — классический брокер сообщений, реализующий протокол AMQP (Advanced Message Queuing Protocol). В отличие от Kafka, RabbitMQ ориентирован на «умную» маршрутизацию и гибкие паттерны доставки.

4.1 Архитектура

Producer ──▶ Exchange ──binding──▶ Queue ──▶ Consumer

                                   ┌─────────┐
              ┌──── routing_key ──▶│ Queue A │──▶ Consumer 1
              │   "order.created"  └─────────┘
┌──────────┐  │                    
│ Exchange │──┤
│ (topic)  │  │
└──────────┘  │                    ┌─────────┐──▶ Consumer 2
              └──── routing_key ──▶│ Queue B │
                     "order.*"     └─────────┘──▶ Consumer 3
                                   

Exchange — маршрутизатор сообщений. Получает сообщения от producer и направляет их в одну или несколько очередей на основе правил (bindings).

Queue — буфер, хранящий сообщения до их обработки. В отличие от Kafka, сообщение удаляется из очереди после acknowledgment.

Binding — правило, связывающее exchange с queue. Определяет, какие сообщения попадают в какую очередь.

4.2 Типы Exchange

Direct Exchange: Сообщение маршрутизируется в очередь, у которой binding key точно совпадает с routing key сообщения.

Producer: publish(routing_key="payment.success", body=...)

Exchange (direct):
  binding: "payment.success" → Queue "payment-notifications"
  binding: "payment.failed"  → Queue "payment-errors"

Fanout Exchange: Сообщение отправляется во все привязанные очереди, routing key игнорируется. Аналог broadcast.

Exchange (fanout) ──▶ Queue "email-notifications"
                  ──▶ Queue "push-notifications"
                  ──▶ Queue "analytics"

Topic Exchange: Маршрутизация по шаблону routing key. Поддерживает wildcards: * (одно слово), # (ноль или более слов).

Producer: publish(routing_key="order.eu.created", body=...)

Exchange (topic):
  binding: "order.*.created"  → Queue A (получит: любой регион, created)
  binding: "order.eu.#"       → Queue B (получит: все события EU)
  binding: "#"                → Queue C (получит: все сообщения)

Headers Exchange: Маршрутизация по заголовкам сообщения, а не по routing key. Полезен для сложной маршрутизации на основе метаданных.

4.3 Подтверждения и надёжность

RabbitMQ поддерживает несколько механизмов обеспечения надёжности:

Publisher Confirms:

channel.confirm_delivery()

try:
    channel.basic_publish(
        exchange="orders",
        routing_key="order.created",
        body=json.dumps(order),
        properties=pika.BasicProperties(
            delivery_mode=2,  # persistent message
        )
    )
    print("Message confirmed")
except pika.exceptions.UnroutableError:
    print("Message was returned (no matching queue)")

Consumer Acknowledgments:

def callback(ch, method, properties, body):
    try:
        process_message(body)
        ch.basic_ack(delivery_tag=method.delivery_tag)  # Успех
    except Exception:
        ch.basic_nack(
            delivery_tag=method.delivery_tag,
            requeue=False  # Не возвращать в очередь, отправить в DLQ
        )

channel.basic_consume(queue="orders", on_message_callback=callback)

Persistent Messages + Durable Queues:

• delivery_mode=2 — сообщение записывается на диск
• durable=True — очередь пережившает рестарт брокера
• Комбинация обеих настроек минимизирует потерю данных (но не исключает полностью — данные могут быть в page cache ОС и не сброшены на диск)

4.4 RabbitMQ vs Kafka

Когда выбирать RabbitMQ:

• Сложная маршрутизация сообщений (topic exchange, headers)
• Нужны приоритеты сообщений
• Нужен низкий latency при небольших объёмах
• Task queues (Celery, background jobs)
• Протоколы MQTT/STOMP (IoT)

Когда выбирать Kafka:

• Высокий throughput (сотни тысяч–миллионы msg/s)
• Нужен replay (повторное чтение событий)
• Event sourcing, CDC, stream processing
• Множество независимых consumer groups
• Долгосрочное хранение событий

---

5 Amazon SQS/SNS

Облачные managed-сервисы от AWS, которые избавляют от необходимости управлять инфраструктурой.

5.1 Amazon SQS (Simple Queue Service)

Полностью управляемая очередь сообщений. Два типа:

Standard Queue:

• Почти неограниченный throughput
• At-least-once delivery (возможны дубликаты)
• Best-effort ordering (порядок не гарантирован)
• Сообщение может быть доставлено более одного раза

FIFO Queue:

• До 3000 msg/s (с batching до 30000 msg/s)
• Exactly-once processing (через deduplication ID)
• Строгий порядок в рамках Message Group ID
• Дороже Standard

import boto3

sqs = boto3.client("sqs")

# Отправка
sqs.send_message(
    QueueUrl="https://sqs.us-east-1.amazonaws.com/123/orders",
    MessageBody=json.dumps({"order_id": "abc", "amount": 99.99}),
    MessageGroupId="user-123",         # Только для FIFO
    MessageDeduplicationId="order-abc", # Только для FIFO
)

# Получение
response = sqs.receive_message(
    QueueUrl="https://sqs.us-east-1.amazonaws.com/123/orders",
    MaxNumberOfMessages=10,
    WaitTimeSeconds=20,  # Long polling
)

for msg in response.get("Messages", []):
    process(msg["Body"])
    sqs.delete_message(
        QueueUrl="https://sqs.us-east-1.amazonaws.com/123/orders",
        ReceiptHandle=msg["ReceiptHandle"],
    )

Visibility Timeout: После получения сообщения оно становится «невидимым» для других consumer-ов на заданное время. Если consumer не подтвердит обработку (delete), сообщение станет доступным повторно.

5.2 Amazon SNS (Simple Notification Service)

Managed Pub/Sub сервис. Producer публикует сообщение в topic, и оно доставляется всем подписчикам.

Подписчики SNS:

• SQS queue
• Lambda function
• HTTP/HTTPS endpoint
• Email
• SMS
• Mobile push (APNs, FCM)

Типичный паттерн — SNS + SQS (Fan-out):

                         ┌──── SQS "payment-queue" ──▶ Payment Service
                         │
Order Service ──▶ SNS ───┼──── SQS "shipping-queue" ──▶ Shipping Service
 "order.created"         │
                         ├──── SQS "analytics-queue" ──▶ Analytics Service
                         │
                         └──── Lambda ──▶ Send confirmation email

Каждый сервис имеет свою SQS-очередь, подписанную на SNS-топик. Это обеспечивает:

• Независимость: Каждый сервис обрабатывает сообщения в своём темпе
• Надёжность: Если один сервис упал, остальные продолжают работу
• Гарантии: SQS обеспечивает retry и DLQ для каждого подписчика

5.3 SQS/SNS vs Kafka vs RabbitMQ

---

6 Гарантии доставки

Гарантия доставки — одна из центральных тем при работе с очередями. Это контракт между producer, broker и consumer о том, сколько раз сообщение будет обработано.

6.1 At-Most-Once (Не более одного раза)

Сообщение может быть потеряно, но никогда не будет обработано дважды.

Producer ──send──▶ Broker ──deliver──▶ Consumer
         (no retry)        (ack before processing)

Реализация:

• Producer: acks=0, без retry
• Consumer: фиксирует offset до обработки сообщения

# Consumer: commit offset сразу, потом обрабатываем
consumer.commit()  # offset зафиксирован
process(message)    # Если упадёт здесь — сообщение потеряно

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

6.2 At-Least-Once (Не менее одного раза)

Сообщение гарантированно будет обработано, но может быть обработано более одного раза.

Producer ──send──▶ Broker ──deliver──▶ Consumer
        (retry on failure)  (ack after processing)

Реализация:

• Producer: acks=all, с retry
• Consumer: фиксирует offset после успешной обработки

process(message)    # Обрабатываем
consumer.commit()   # Фиксируем offset
# Если упадёт между process и commit → при рестарте сообщение обработается повторно

Проблема дублирования: Если consumer обработал сообщение, но упал до фиксации offset — при перезапуске он обработает его снова. Поэтому at-least-once требует идемпотентной обработки.

def process_payment(message):
    payment_id = message["payment_id"]

    # Идемпотентность: проверяем, не обработали ли мы это уже
    if db.exists("processed_payments", payment_id):
        return  # Уже обработано — пропускаем

    db.execute_payment(payment_id, message["amount"])
    db.insert("processed_payments", payment_id)

Когда использовать: Подавляющее большинство бизнес-сценариев. Платежи, заказы, уведомления — с идемпотентной обработкой.

6.3 Exactly-Once (Ровно один раз)

Самая сложная и ресурсоёмкая гарантия. Каждое сообщение обрабатывается гарантированно один раз — без потерь и без дублирования.

Почему это сложно: Exactly-once в распределённой системе — это, по сути, распределённая транзакция между producer, broker и consumer. Невозможно гарантировать exactly-once в общем случае (из-за проблемы двух генералов), но можно добиться effectively exactly-once через комбинацию механизмов.

Kafka Exactly-Once Semantics (EOS):

1. Idempotent Producer: Kafka присваивает каждому producer уникальный Producer ID и каждому сообщению Sequence Number. Если producer повторно отправляет сообщение (из-за retry), broker распознаёт дубликат по PID + SeqNum и игнорирует его.

producer = KafkaProducer(
    enable_idempotence=True,  # Включаем идемпотентность
    acks="all",
)

2. Transactional API: Позволяет атомарно записать сообщения в несколько партиций/топиков и зафиксировать consumer offset в одной транзакции.

producer.init_transactions()

try:
    producer.begin_transaction()

    # Записываем результат обработки
    producer.send("output-topic", key=key, value=result)

    # Фиксируем offset входного сообщения
    producer.send_offsets_to_transaction(
        offsets={TopicPartition("input-topic", 0): OffsetAndMetadata(42)},
        group_id="my-consumer-group",
    )

    producer.commit_transaction()
except Exception:
    producer.abort_transaction()

Либо все операции (запись результата + фиксация offset) происходят атомарно, либо ни одна.

Когда использовать: Финансовые транзакции, агрегация метрик (подсчёт кликов, расходов), ETL-пайплайны — где дублирование или потеря неприемлемы. Стоимость: повышенная латенси (~5-10x по сравнению с at-least-once), сниженный throughput.

Сравнительная таблица

Практический совет: В большинстве систем at-least-once + idempotent consumer — оптимальный выбор. Это проще, быстрее и надёжнее, чем exactly-once, и при правильной реализации идемпотентности даёт тот же бизнес-результат.

---

7 Dead Letter Queue и retry-стратегии

7.1 Что происходит при ошибке обработки?

Ошибки обработки делятся на два класса:

Transient errors (временные): Сбои сети, таймаут БД, rate limit от внешнего API. Повторная попытка с высокой вероятностью завершится успехом.

Permanent errors (постоянные): Невалидный формат сообщения, бизнес-ошибка (пользователь не существует), баг в обработчике. Повторные попытки бесполезны.

7.2 Retry Queue

Для временных ошибок сообщение отправляется в retry queue с задержкой перед повторной обработкой.

Main Queue ──▶ Consumer ──(ошибка)──▶ Retry Queue (delay: 30s)
                  ▲                          │
                  │                          │
                  └──────────────────────────┘
                     (через 30 секунд)

Многоуровневый retry с возрастающей задержкой:

Main Queue → Consumer → (fail) → Retry-1 (delay: 1 min)
                                 → (fail) → Retry-2 (delay: 5 min)
                                            → (fail) → Retry-3 (delay: 30 min)
                                                        → (fail) → DLQ

7.3 Dead Letter Queue (DLQ)

DLQ — очередь для сообщений, которые не удалось обработать после всех попыток. Это «карантин» для проблемных сообщений, позволяющий:

• Изолировать проблему — не блокировать основную очередь
• Сохранить данные — сообщения не теряются
• Отложить решение — инженер может разобраться позже и переотправить

MAX_RETRIES = 3

def process_with_retry(message):
    retry_count = message.headers.get("x-retry-count", 0)

    try:
        process(message)
        ack(message)
    except TransientError:
        if retry_count < MAX_RETRIES:
            # Отправляем в retry queue с увеличенным счётчиком
            publish_to_retry_queue(
                message,
                headers={"x-retry-count": retry_count + 1},
                delay=exponential_backoff(retry_count),
            )
        else:
            # Исчерпали попытки — отправляем в DLQ
            publish_to_dlq(message, reason="Max retries exceeded")
        ack(message)  # Убираем из основной очереди
    except PermanentError as e:
        # Повторять бессмысленно — сразу в DLQ
        publish_to_dlq(message, reason=str(e))
        ack(message)

DLQ в RabbitMQ:

# Основная очередь с DLQ
channel.queue_declare(
    queue="orders",
    arguments={
        "x-dead-letter-exchange": "dlx",          # DLX exchange
        "x-dead-letter-routing-key": "orders.dlq", # Routing key для DLQ
        "x-message-ttl": 60000,                    # TTL для retry (60s)
    }
)

# DLQ
channel.queue_declare(queue="orders.dlq")
channel.queue_bind(
    queue="orders.dlq",
    exchange="dlx",
    routing_key="orders.dlq",
)

DLQ в Amazon SQS:

sqs.create_queue(
    QueueName="orders",
    Attributes={
        "RedrivePolicy": json.dumps({
            "deadLetterTargetArn": "arn:aws:sqs:us-east-1:123:orders-dlq",
            "maxReceiveCount": "3",  # После 3 неудачных попыток → DLQ
        })
    }
)

7.4 Retry-стратегии

Immediate Retry: Повторная попытка немедленно. Простейший подход, но опасен: если ошибка из-за перегрузки downstream-сервиса, мгновенные ретраи усугубят проблему.

Fixed Delay: Повторная попытка через фиксированный интервал (например, 5 секунд). Лучше immediate retry, но может создавать пики нагрузки (все ретраи происходят одновременно).

Exponential Backoff: Интервал между попытками растёт экспоненциально: 1s → 2s → 4s → 8s → 16s.

def exponential_backoff(retry_count: int, base: float = 1.0, max_delay: float = 60.0) -> float:
    delay = min(base * (2 ** retry_count), max_delay)
    return delay

Exponential Backoff + Jitter: К exponential backoff добавляется случайный разброс, чтобы предотвратить retry storm (thundering herd при ретраях) — ситуацию, когда множество клиентов, получивших ошибку одновременно, ретраят в одно и то же время.

import random

def backoff_with_jitter(retry_count: int, base: float = 1.0, max_delay: float = 60.0) -> float:
    delay = min(base * (2 ** retry_count), max_delay)
    jitter = random.uniform(0, delay)  # Full jitter
    return jitter

# Retry 0: random(0, 1s)
# Retry 1: random(0, 2s)
# Retry 2: random(0, 4s)
# Retry 3: random(0, 8s)

Три варианта jitter:

• Full Jitter: random(0, delay) — максимальное распределение
• Equal Jitter: delay/2 + random(0, delay/2) — гарантированный минимальный delay
• Decorrelated Jitter: min(max_delay, random(base, prev_delay * 3)) — AWS рекомендует как лучший вариант

---

8 Event-Driven Architecture: Event Sourcing и CQRS

8.1 Event Sourcing

Традиционный подход хранит текущее состояние объекта (баланс кошелька: 500₽). Event Sourcing хранит последовательность всех событий, которые привели к этому состоянию.

Традиционный подход:
  wallet: {user_id: 123, balance: 500}

Event Sourcing:
  event1: {type: "deposited", amount: 1000, timestamp: T1}
  event2: {type: "withdrawn", amount: 300, timestamp: T2}
  event3: {type: "withdrawn", amount: 200, timestamp: T3}
  → Текущий баланс = 1000 - 300 - 200 = 500

Kafka как event store: Команды (запросы на изменение) помещаются в Kafka-топик. Consumer читает их, валидирует, преобразует в события и применяет к состоянию.

Command: "withdraw 200₽ from wallet 123"
     │
     ▼
┌─────────┐    ┌──────────────┐    ┌─────────────┐
│ Command │    │ Event Store  │    │ Read Model  │
│ Handler │──▶ │ (Kafka)      │──▶ │ (PostgreSQL)│
│         │    │ append event │    │ materialized│
└─────────┘    └──────────────┘    │ view        │
                                   └─────────────┘

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

• Полный аудит: Каждое изменение зафиксировано. Идеально для финансовых систем, медицины, compliance
• Воспроизводимость: Можно восстановить состояние на любой момент времени, «проиграв» события заново
• Отладка: Если баланс некорректен, можно посмотреть каждое событие, приведшее к нему
• Temporal queries: «Какой баланс был 15 января в 14:30?»

Недостатки:

• Сложность: Значительно сложнее CRUD в разработке и поддержке
• Eventual consistency: Состояние read-модели может отставать от event store
• Schema evolution: Изменение формата событий требует версионирования
• Storage: Хранение всей истории может быть дорогим. Решение — snapshots: периодическое сохранение текущего состояния + replay только с последнего snapshot

8.2 CQRS (Command Query Responsibility Segregation)

CQRS — разделение моделей записи (commands) и чтения (queries) на разные сервисы с разными хранилищами.

                      ┌───────────────────┐
                      │   Command Side    │
   Write ──────────▶  │   (PostgreSQL)    │
   (create, update,   │  Нормализованная  │
    delete)           │       схема       │
                      └─────────┬─────────┘
                                │ Events
                                ▼
                      ┌───────────────────┐
                      │     Event Bus     │
                      │      (Kafka)      │
                      └─────────┬─────────┘
                                │
                                ▼
                      ┌────────────────────┐
   Read ◀──────────── │     Query Side     │
   (search, list,     │  (Elasticsearch,   │
    aggregate)        │Redis, denormalized │
                      │    PostgreSQL)     │
                      └────────────────────┘

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

• Независимое масштабирование: Чтение и запись масштабируются отдельно (обычно чтений больше)
• Оптимизированные модели: Write-модель нормализована (целостность), read-модель денормализована (скорость)
• Разные хранилища: PostgreSQL для записи, Elasticsearch для полнотекстового поиска, Redis для dashboards

Недостатки:

• Eventual consistency: Read-модель обновляется асинхронно, с задержкой
• Сложность: Два хранилища, синхронизация, обработка ошибок

Когда оправдано: Системы с высокой асимметрией чтения/записи, сложными запросами чтения и необходимостью разных представлений данных (лента, поиск, аналитика).

---

9 Backpressure

Backpressure — механизм управления потоком, при котором получатель сигнализирует отправителю о необходимости снизить скорость, чтобы избежать перегрузки.

Почему это важно

Если producer генерирует 100K msg/s, а consumer обрабатывает 10K msg/s, очередь будет расти бесконечно. Без backpressure это приводит к:

• Исчерпанию памяти/диска брокера
• Росту латенси (сообщения ждут в очереди всё дольше)
• Каскадным отказам

Механизмы backpressure

Pull-модель (Kafka): Consumer сам вытягивает данные, когда готов их обработать. Это встроенный backpressure: если consumer медленный, он просто забирает меньше данных, а сообщения накапливаются в Kafka (которая оптимизирована для хранения на диске).

# Consumer контролирует скорость
records = consumer.poll(timeout_ms=1000, max_records=100)
# Если consumer перегружен — он уменьшает max_records или увеличивает timeout

Push-модель с credit-based flow control (RabbitMQ):

# QoS: consumer получает максимум N необработанных сообщений
channel.basic_qos(prefetch_count=10)
# Broker не отправит 11-е сообщение, пока consumer не подтвердит одно из предыдущих

Буферизация (брокер как буфер): Брокер сообщений сам по себе является буфером, поглощающим пики нагрузки. Producer продолжает писать в нормальном режиме, а consumer обрабатывает в своём темпе. Это работает для краткосрочных пиков, но не для длительного дисбаланса скоростей.

Reactive Streams: Спецификация для асинхронной обработки потоков с backpressure (RxJava, Project Reactor, Akka Streams). Consumer запрашивает N элементов, producer отправляет ровно N:

subscription.request(10);  // "Готов обработать 10 элементов"

---

10 Реальные примеры из индустрии

LinkedIn: Kafka для data pipeline

LinkedIn создал Kafka для решения собственных проблем: сбор логов, метрик и user activity events из сотен сервисов. Сегодня LinkedIn обрабатывает триллионы сообщений через Kafka ежедневно. Также LinkedIn разработал Databus — систему CDC (Change Data Capture) для синхронизации изменений между хранилищами данных.

Uber: Kafka + Flink + Pinot

Uber использует Kafka в нескольких критичных пайплайнах:

• Платёжная система: Kafka обеспечивает надёжную доставку платёжных событий между микросервисами
• Рекламная платформа: Связка Kafka → Apache Flink (stream processing) → Apache Pinot (real-time OLAP) для обработки рекламных событий в реальном времени
• Surge pricing: Real-time агрегация спроса и предложения через Kafka Streams

Netflix: Event-driven микросервисы

Netflix использует event-driven архитектуру для развязки сотен микросервисов:

• Fan-out запросы: Один запрос пользователя порождает десятки параллельных запросов к дочерним сервисам
• Hystrix (ныне Resilience4j): Библиотека для circuit breaker и bulkhead, защищающая от каскадных отказов при работе с очередями
• Data pipeline: Kafka для сбора viewing history, рекомендаций и A/B тестирования

Waze: собственная система очередей

Waze столкнулись с ситуацией, когда стандартные продукты не справлялись с требуемой нагрузкой (десятки тысяч сообщений в секунду с жёсткими требованиями по латенси). Они разработали собственную систему очередей, оптимизированную под их специфичные паттерны доступа.

Facebook: Wormhole и TAO

Facebook использует Wormhole — систему Pub/Sub для надёжной георепликации данных между дата-центрами. В комбинации с TAO (кэширующий слой для social graph) это обеспечивает глобальную консистентность данных.

---

11 Выбор технологии: дерево решений

Нужна ли вам очередь сообщений?
│
├── Синхронный REST/gRPC достаточен → НЕТ
│   (малое число сервисов, низкая нагрузка, нужна мгновенная обратная связь)
│
└── ДА (асинхронность, развязка, пики нагрузки)
    │
    ├── Хотите managed-сервис (минимум ops)?
    │   ├── AWS → SQS/SNS (простые задачи) или Amazon MSK (Kafka managed)
    │   ├── GCP → Cloud Pub/Sub
    │   └── Azure → Azure Service Bus или Azure Event Hubs
    │
    └── Self-hosted?
        │
        ├── Основной use case — task queue / job processing?
        │   └── RabbitMQ (или Redis Streams для простых случаев)
        │
        ├── Основной use case — event streaming / data pipeline?
        │   └── Apache Kafka (или Redpanda как Kafka-compatible альтернатива)
        │
        ├── Нужна сложная маршрутизация (topic, headers)?
        │   └── RabbitMQ
        │
        ├── Нужен replay / event sourcing?
        │   └── Kafka
        │
        ├── Throughput > 100K msg/s?
        │   └── Kafka
        │
        └── Throughput < 10K msg/s, нужна простота?
            └── RabbitMQ или Redis Streams

---

12 Практические рекомендации

Чек-лист при проектировании

1. Определите паттерн: P2P, Pub/Sub или гибрид? Это определит выбор технологии.

2. Выберите гарантию доставки: At-least-once + idempotent consumer — оптимальный выбор для 90% случаев.

3. Спроектируйте идемпотентность: Каждый consumer должен корректно обрабатывать дубликаты. Используйте unique ID сообщения + deduplication table.

4. Определите ordering requirements: Нужен ли строгий порядок? Если да — используйте один partition/queue на entity (все события пользователя в одну партицию).

5. Спроектируйте retry + DLQ: Exponential backoff с jitter для retry. DLQ для безнадёжных сообщений. Мониторинг DLQ с алертами.

6. Определите размер сообщения: Kafka оптимален для сообщений до 1 MB. Для больших payload-ов — храните данные в S3/MinIO, а в сообщении передавайте ссылку (claim check pattern).

7. Мониторинг:

   • Consumer lag — разница между последним записанным и последним обработанным offset. Растущий lag = consumer не успевает.
   • Throughput (msg/s in, msg/s out)
   • Error rate (failed processing / total)
   • DLQ depth — количество сообщений в DLQ
   • End-to-end latency — время от отправки до обработки

8. Планируйте partitioning: Выбирайте ключ партиционирования так, чтобы нагрузка распределялась равномерно. Горячие ключи (один пользователь генерирует 90% событий) — проблема.

9. Версионирование схемы: Используйте Schema Registry (Avro, Protobuf) для эволюции формата сообщений без breaking changes.

---

13 Вопросы для самопроверки

1. Вы проектируете систему обработки заказов в e-commerce. При создании заказа нужно: списать деньги, зарезервировать товар на складе, отправить уведомление пользователю. Как вы организуете коммуникацию между сервисами? Обоснуйте выбор между синхронным и асинхронным подходом.

2. Объясните разницу между acks=0, acks=1 и acks=all в Kafka. Для каких use cases подходит каждый вариант?

3. В Kafka-топике 12 партиций, а consumer group состоит из 15 consumer-ов. Что произойдёт? Как это исправить?

4. Ваша система использует at-least-once delivery. Consumer обрабатывает платежи. Как реализовать идемпотентность, чтобы платёж не был проведён дважды?

5. Сравните RabbitMQ и Kafka для задачи: обработка 500K событий клик-стрима в секунду от 10 независимых аналитических сервисов. Обоснуйте выбор.

6. Что такое consumer lag в Kafka? Как его мониторить и какие действия предпринять, если lag растёт?

7. Ваш producer отправляет сообщения в Kafka со скоростью 200K msg/s, но один из consumer-ов обрабатывает только 50K msg/s. Какие механизмы backpressure вы примените?

8. Объясните паттерн SNS + SQS fan-out. В каких ситуациях он предпочтительнее direct Pub/Sub?

9. Ваша DLQ начала быстро заполняться. Опишите процесс анализа и решения проблемы.

10. Сравните Event Sourcing + CQRS с классическим CRUD-подходом. В каких системах Event Sourcing оправдан, а в каких — избыточен?

---

14 Дополнительные ресурсы

Papers

• Kafka: a Distributed Messaging System for Log Processing (Kreps, Narkhede, Rao, 2011) — оригинальная статья LinkedIn о создании Kafka. Описывает мотивацию, архитектуру и первые результаты.
• The Log: What every software engineer should know about real-time data's unifying abstraction (Jay Kreps, 2013) — фундаментальное эссе о commit log как универсальной абстракции для данных. Обязательно к прочтению.
• Building LinkedIn's Real-time Activity Data Pipeline (Goodhope et al., 2012) — практический опыт масштабирования data pipeline в LinkedIn.
• Wormhole: Reliable Pub-Sub to Support Geo-replicated Internet Services (Sharma et al., Facebook, 2015) — Pub/Sub для георепликации в Facebook.

Блоги

• Confluent Blog (confluent.io/blog) — статьи от создателей Kafka: best practices, новые фичи, архитектурные решения.
• RabbitMQ Blog (blog.rabbitmq.com) — официальный блог RabbitMQ с разборами паттернов и оптимизаций.
• Uber Engineering: Reliable Reprocessing — как Uber строит retry-стратегии для Kafka consumers.
• AWS Architecture Blog: Fan-out pattern with SNS and SQS — паттерн fan-out с примерами и best practices.

Talks

• "Turning the Database Inside-Out" (Martin Kleppmann, Strange Loop) — как event log меняет архитектуру систем. Одно из лучших выступлений по event-driven архитектуре.
• "ETL Is Dead, Long Live Streams" (Neha Narkhede, QCon) — эволюция от batch ETL к stream processing с Kafka.
• "Four Distributed Systems Architectural Patterns" (Tim Berglund, Confluent) — обзор паттернов event-driven архитектуры.
• "The Many Meanings of Event-Driven Architecture" (Martin Fowler, GOTO) — разбор терминологии и типичных заблуждений.