Спроектируйте распределённую очередь сообщений

Требования

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

Функциональные

• Произвести (produce): дописать сообщение в именованный топик.
• Потребить (consume): читать сообщения из топика, по порядку, начиная с выбранной позиции.
• Множество независимых консьюмеров: несколько разных систем читают один топик, не мешая друг другу (скоринг фрода, бухгалтерия, аналитический пайплайн — все читают поток платежей).
• Реплей: консьюмер может перемотать назад и перечитать прошлые сообщения (пропавшая фича из хука).
• Удержание (retention): сообщения хранятся настроенное окно (время или размер), а не удаляются при доставке.

Нефункциональные

• Высокая пропускная способность. Миллионы сообщений/с — это пожарный шланг, не почтовый ящик.
• Надёжность. Подтверждённое сообщение обязано пережить крах брокера; потеря платёжного события недопустима.
• Горизонтальный масштаб. Добавь брокеров — добавь мощности; ни один брокер не потолок.
• Гарантия порядка — хотя бы внутри партиции.
• Настраиваемая семантика доставки. Разные консьюмеры терпят разные режимы отказа (пайплайн метрик может уронить сообщение; бухгалтерия — нет).

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

Оценка

• Цель — 1 миллион сообщений/с на пике, среднее сообщение 1 КБ → ~1 ГБ/с пропускной способности записи. Это заголовочное число, и оно исключает любой дизайн на одном брокере.
• Удержание 7 дней: 1 ГБ/с × 86 400 с × 7 ≈ ~600 ТБ лога на диске (до репликации). При факторе репликации 3 — ~1,8 ПБ. Лог живёт на диске, не в RAM — так что последовательный дисковый I/O и есть вся игра.
• Пропускная способность на партицию упирается, может, в ~10 МБ/с для упорядоченных реплицированных записей, так что 1 ГБ/с требует порядка ~100+ партиций, разбросанных по брокерам — число партиций и есть то, чем покупают пропускную способность.
• Консьюмеры: если чтения в основном последовательны с хвоста лога, page cache ОС обслуживает свежие сообщения из RAM, так что здоровый кластер почти не делает случайных дисковых чтений — паттерн доступа и есть то, что заставляет числа работать.

Ключевая мысль из чисел: последовательная дозапись на диск плюс партиционирование — то, что превращает «медленный» диск в пожарный шланг на 1 ГБ/с, ведь последовательная пропускная способность диска соперничает с памятью, тогда как случайный I/O был бы в ~1000 раз медленнее.

Высокоуровневый дизайн

Топик делится на партиции; каждая партиция — упорядоченный лог только на дозапись, живущий на брокере-лидере и реплицируемый на брокеры-фолловеры. Продюсеры дописывают в партиции (по ключу или round-robin). Консьюмеры в группе консьюмеров делят партиции между собой и отслеживают оффсет на партицию.

Глубокое погружение

Партиции: единица параллелизма и порядка

Топик делится на партиции, и партиция — атом всей системы. Из неё следуют три вещи:

• Параллелизм. Разные партиции живут на разных брокерах и пишутся/читаются независимо, так что пропускная способность масштабируется с числом партиций — так доходишь от лимита одного брокера до 1 ГБ/с.
• Порядок — на партицию, не на топик. Сообщения внутри партиции строго упорядочены по оффсету; между партициями нет глобального порядка. Так что если нужна упорядоченная обработка событий пользователя, надо маршрутизировать все его сообщения в одну и ту же партицию — делается хешированием ключа партиции (например user_id). Нет ключа → round-robin → максимальный разброс, но нет гарантии порядка.
• Параллелизм консьюмеров ограничен партициями. Внутри группы консьюмеров каждую партицию потребляет ровно один консьюмер, так что активных консьюмеров может быть не больше, чем партиций. Число партиций — это и ручка пропускной способности, и максимум параллелизма консьюмеров — выбирается наперёд и неловко меняется позже.

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

Репликация и ISR

У каждой партиции один лидер (берёт все чтения и записи) и несколько фолловеров, реплицирующих лог лидера. Надёжность держится на множестве in-sync реплик (ISR): фолловеров, что сейчас догнали лидера. Запись считается закоммиченной лишь когда все члены ISR её имеют, а acks управляет строгостью продюсера:

• acks=0 — выстрелил-и-забыл; быстрее всего, может потерять данные.
• acks=1 — подтверждение лишь лидера; теряет данные, если лидер умер до репликации на фолловеров.
• acks=all — каждый член ISR должен иметь сообщение до подтверждения; переживает отказ лидера без потерь, пока ISR не пуст.

Когда лидер умирает, новый лидер избирается из ISR, так что у него уже есть каждое закоммиченное сообщение — ни одна закоммиченная запись не теряется. Настраиваемый min.insync.replicas задаёт, сколько реплик должны быть in-sync для прохода записи acks=all; поставь 2 (при факторе репликации 3) — и переживёшь отказ одного брокера, всё ещё отклоняя записи, если надёжность была бы скомпрометирована.

Оффсеты, группы консьюмеров и семантика доставки

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

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

Но когда ты коммитишь оффсет относительно того, когда обрабатываешь сообщение, и определяет семантику доставки — важнейшее решение корректности в любой async-системе:

• At-most-once: коммитить оффсет до обработки. Если консьюмер упал посреди обработки, сообщение пропускается при перезапуске — возможна потеря, нет дублей. (Приемлемо для метрик.)
• At-least-once: коммитить оффсет после обработки. Если консьюмер упал после обработки, но до коммита, сообщение перечитывается и переобрабатывается — возможны дубли, нет потерь. (Обычный дефолт.)
• Exactly-once: ни дублей, ни потерь. По-настоящему сложно — требует либо идемпотентной обработки (консьюмер дедуплицирует по ID сообщения, так что переобработка безвредна), либо транзакции, атомарно коммитящей оффсет и побочный эффект вместе. Kafka предлагает транзакционных продюсеров/консьюмеров для этого, но большинство систем добиваются «фактически раз» через at-least-once доставку плюс идемпотентный консьюмер.

Узкие места и компромиссы

• Число партиций — почти-односторонее решение. Слишком мало — упрёшься в пропускную способность и параллелизм консьюмеров; слишком много — платишь координацией, файловыми дескрипторами и ребалансами. Увеличение партиций позже ломает порядок по ключу (ключ может хешироваться в новую партицию), поэтому размеряешь щедро наперёд.
• Горячая партиция / перекошенный ключ. Плохой ключ партиции (например по стране, где одна доминирует) шлёт почти весь трафик в одну партицию и одному консьюмеру — версия горячего шарда для лога, и весь кластер помочь не может.
• At-least-once значит, что консьюмеры обязаны быть идемпотентными. Обычный дефолт отдаёт дубли при ретрае, так что каждый консьюмер с побочными эффектами (списать карту, отправить письмо) обязан дедуплицировать по ID сообщения — требование, о котором команды забывают, пока кого-то дважды не спишут.
• Штормы ребалансов. Когда консьюмер входит или выходит из группы, партиции переназначаются и потребление паузится; частая текучка членства (мигающие консьюмеры, долгие паузы GC) вызывает повторные ребалансы, останавливающие группу — так что тюнинг живости консьюмеров операционно важен.
• Удержание против стоимости. Долгое удержание включает реплей, но стоит петабайтов; окно — прямой размен деньги-против-восстановимости, и «бесконечное удержание ради реплея» — реальный счёт.