Консьюмеры брокеров: prefetch, ack, backpressure

Перед началом: модули 6, 7, 11, 13 (конвейеры, пулы, shutdown, очереди в процессе). Инструменты: Go 1.22+, go run, брокер в докере по желанию. Трудоёмкость: ~4 ч (теория 1.5 ч · практика 2 ч · квиз 0.5 ч).

🔥 Инцидент: 10 000 сообщений в секунду, OOM и потерянные заказы

Консьюмер заказов читает топик брокера: 10 000 сообщений в секунду, обработка одного — 5 мс. По арифметике нужно 50 параллельных обработчиков; на деле один поток давал 200 сообщений в секунду, и лаг потребителя рос всю ночь.

Первое «решение» — горутина на сообщение. Через час процесс убил OOM-killer: в памяти висели сотни тысяч горутин, каждая со своим контекстом, телом сообщения и буферами.

Второе «решение» — ack сразу после чтения, чтобы брокер не ждал. Лаг исчез, латентность упала, а при первом же деплое 3 000 заказов испарились: брокер считал их доставленными, приложение — ещё нет.

Оба решения лечили симптом. Настоящих рычагов три: prefetch-окно (сколько сообщений держать в работе), ack после обработки и идемпотентность (чтобы повторная доставка была безопасной).

📖 Теория

Брокер и приложение договариваются о трёх вещах: сколько сообщений приложение берёт «в работу», когда подтверждает доставку и что делает с теми, кого не смогло обработать. Всё остальное — детали транспорта.

16.1 Конкурентные воркеры на канале

Читать канал брокера из нескольких горутин «как получится» нельзя: порядок внутри партиции теряется, а коммит offset'а становится гонкой. Правильная схема — один читатель, который раскладывает сообщения в канал, и пул воркеров, которые его разбирают:

go
func consume(in <-chan Msg, workers int, handle func(Msg) error) {
    jobs := make(chan Msg)
    var wg sync.WaitGroup
    for i := 0; i < workers; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for m := range jobs {
                _ = handle(m)
            }
        }()
    }
    for m := range in {
        jobs <- m
    }
    close(jobs)
    wg.Wait()
}

Число воркеров считается по бюджету: workers = throughput × latency. 10 000 msg/s × 5 мс = 50 обработчиков одновременно. Меньше — лаг растёт; больше — пул соединений к БД и внешним API становится узким местом (модуль 7). Воркер на сообщение без пула — это и был OOM.

16.2 Prefetch-окно = bounded-канал

jobs выше небуферизован: читатель ждёт, пока воркер не возьмёт сообщение. Это честно, но дорого — каждый цикл чтения упирается в свободного воркера. Буфер канала становится prefetch-окном: сколько сообщений приложение готово держать «в полёте», ещё не подтвердив их.

go
jobs := make(chan Msg, prefetch) // N незакоммиченных сообщений

Окно — не «ускорение», а долг перед брокером: эти сообщения уже не у брокера, но ещё не ack'нуты. Упал процесс — брокер должен переотдать всё окно. Поэтому размер окна — это одновременно:

запас сглаживания (пики не блокируют читателя);
объём повторной доставки после падения;
верхняя граница памяти на сообщения (модуль 6).

RabbitMQ называет это prefetch count, Kafka — max.poll.records плюс fetch.max.bytes, Redis Streams — COUNT в XREADGROUP. Смысл один: ограничить число незавершённых доставок.

16.3 Ack после обработки: at-least-once и идемпотентность

Порядок операций — единственное, что определяет гарантии:

Порядок	Гарантия	Риск
ack → обработка	at-most-once	падение = потеря
обработка → ack	at-least-once	падение = дубль
обработка + ack в TX	effectively-once	сложно и дорого

At-most-once выбирают для метрик и логов, где потеря дешевле дубля. At-least-once — для денег и заказов, но тогда обработчик обязан быть идемпотентным: повторное сообщение не должно списать деньги дважды.

Идемпотентность — это не «проверить, было ли», а атомарный ключ: INSERT ... ON CONFLICT DO NOTHING по message_id, условный UPDATE по версии, таблица обработанных ключей в той же транзакции, что и эффект. Проверка «сначала SELECT, потом INSERT» — гонка, дубль пройдёт.

16.4 Пауза консьюмера при полном окне

Когда окно заполнено, читатель перестаёт забирать сообщения у брокера. Это и есть backpressure на уровне протокола, и у каждого брокера свой способ его включить:

Kafka: consumer.Pause(partitions) / Resume, плюс ограничение max.poll.records; не вызываете Poll — нет и новых записей.
RabbitMQ: prefetch count на канале; брокер сам не отдаёт больше N неподтверждённых сообщений.
Redis Streams: XREADGROUP ... COUNT N; непрочитанные остаются в PEL и переотдаются через XAUTOCLAIM.
NATS JetStream: MaxAckPending и явный Ack().

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

16.5 Poison message, ретраи и DLQ

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

Рабочая схема:

Обработка упала — вернуть сообщение с задержкой (backoff), а не сразу: мгновенный ретрай крутит цикл на месте.
Счётчик попыток хранится в заголовке или в отдельном хранилище.
После N попыток — DLQ (dead letter queue): отдельный топик для разбора людьми, а не мусорка.
В DLQ сообщение подтверждается, иначе оно снова и снова всплывёт в исходной очереди.

В коде ниже ядовитое сообщение тоже ack'ается — после отправки в DLQ, иначе консьюмер зациклится на нём навсегда.

16.6 Компромиссы

Окно больше vs меньше: большое окно сглаживает пики и держит воркеров занятыми, но увеличивает переотдачу после падения и память; маленькое окно честнее показывает узкое место.
Ack раньше vs позже: раньше — меньше дублей и лага, но потеря при падении; позже — at-least-once, но нужна идемпотентность.
Воркеров больше vs меньше: больше — выше параллелизм, но давление на БД и внешние API; лишние воркеры стоят на канале.
Ретраи в консьюмере vs очередь ретраев: в консьюмере проще, но задерживает партицию; отдельная очередь с TTL возвращает сообщение по расписанию и не блокирует поток.
DLQ автоматически vs вручную: автоматика не теряет сообщения, но может спрятать баг; ручной разбор дороже, зато виден.

💻 Практика

Воспроизведите потерю при раннем ack. Консьюмер с окном 8 и обработчиком, который падает на каждом 5-м сообщении; вариант А — ack сразу после чтения, вариант Б — ack после обработки. Посчитайте, сколько сообщений не дошло до «базы» в каждом варианте.
Настройте окно под бюджет. Обработка 5 мс, цель 2 000 msg/s. Подберите число воркеров и размер окна, замерьте пиковую память и лаг при окнах 0, 16, 256, 4096; объясните разницу.
Сделайте ретраи с backoff и DLQ. Ядовитое сообщение (кратное 13) проходит 3 попытки с задержкой 10/50/200 мс, затем уходит в DLQ; проверьте, что остальные сообщения его не ждут.
Запустите сниппет модуля: брокер на 30 сообщений, окно 4, 3 воркера, ядовитые — кратные 13. Проверьте, что ack'нуто 28 из 30, а в DLQ ровно два сообщения.

Фактический вывод:

обработано и закоммичено: 28 из 30
ядовитых в DLQ: 2 [13 26] (кратные 13)
ack после обработки + идемпотентность = at-least-once без дублей

🧩 Quiz

1. Когда консьюмер подтверждает сообщение при at-least-once?

2. Зачем ядовитое сообщение подтверждают после отправки в DLQ?

3. Чем является размер prefetch-окна?

4. Что даёт пауза чтения при полном окне?

5. Почему «SELECT, потом INSERT» не делает обработчик идемпотентным?

✅ Итоги модуля

Студент должен:

Считать число воркеров по throughput × latency
Настраивать prefetch-окно и объяснять его цену
Выбирать порядок ack под нужную гарантию доставки
Делать обработчик идемпотентным атомарным ключом
Включать паузу консьюмера и видеть её в метриках
Отправлять poison message в DLQ с ретраями и backoff
Объяснить компромиссы окна, порядка ack и ретраев
Ответить на все вопросы Quiz

🔗 Мостик к Модулю 17

Консьюмер с окном и пулом воркеров — это десятки горутин, и каждая может зависнуть навсегда: забытый канал ack, слушатель без done, таймер без Stop. Дальше — как найти такие горутины по pprof, метрикам и leak-тестам, пока они не съели память.

← Модуль 15: Кэш, singleflight и thundering herd Модуль 17: Утечки горутин и диагностика →