Очереди в процессе: диспетчер и ретраи

Перед началом: модули 6, 11–12 (конвейеры, shutdown, bulkhead). Инструменты: Go 1.22+, go test -race для проверки диспетчера. Трудоёмкость: ~4 ч (теория 1.5 ч · практика 2 ч · квиз 0.5 ч).

🔥 Инцидент: письмо в горутине, которое никто не получил

Регистрация заканчивается ответом 201, а письмо с подтверждением отправляется «потом»: go sendEmail(user). Быстро, просто, на стенде работает. В прод деплое (модуль 11) письма теряются: процесс умирает раньше, чем горутина дошла до SMTP. В Чёрную пятницу регистрации выросли в десять раз — и 40 тысяч одновременных горутин с шаблонами и вложениями съели память: под получил OOMKilled.

Заказчик формулирует требование без компромиссов: «письмо должно уйти, даже если мы упали сразу после ответа». Вопрос перестаёт быть про горутины и становится про очередь, гарантии и повторную доставку.

📖 Теория

13.1 Bounded-очередь: почему канал, а не слайс

Слайс под мьютексом — это просто массив: положить и забрать можно, но «подождать, пока появится место» и «проснуться, когда придёт задача» придётся писать руками через sync.Cond или каналы. Канал даёт это из коробки: блокировку при полном буфере, пробуждение читателя, select с ctx.Done() и безопасное закрытие.

Главное решение — bound: очередь конечна. Небуферизованная очередь честно передаёт давление назад; буфер на N задач сглаживает всплеск ровно на N. Неограниченная очередь в памяти — это OOM, отложенный до первого медленного дня.

go
queue := make(chan Task, 1000) // bound — часть контракта
metrics.Gauge("queue_len", float64(len(queue)))

len(queue) — приблизительный показатель: он не заменяет семантику, но даёт сигнал, что очередь подошла к пределу.

13.2 Переполнение: блокировать, сбрасывать или отказывать

Три политики, и выбирать нужно осознанно, по классу работы:

  • Блокировать (queue <- t): давление уходит вызывающему. Для HTTP это ожидание в хендлере и риск таймаута; зато задача не теряется.
  • Shed (select + default): при полной очереди новая задача отбрасывается и считается. Годится для аналитики, прогрева, необязательных уведомлений; для писем — обычно нет.
  • Отказ (вернуть ошибку вызывающему): синхронный API отвечает 429/503, и решение принимает вызывающий. Честно, но требует явного контракта.

Блокировка без ctx — ловушка: если очередь встала, хендлер висит до таймаута клиента. Любая блокирующая отправка должна уметь выйти:

go
select {
case queue <- task:
case <-ctx.Done():
    return ctx.Err() // не копим горутины в ожидании
}

Shed не должен быть молчаливым: считайте shed_total и логируйте хотя бы агрегат. Иначе «мы защитили сервис» превращается в «мы незаметно потеряли половину работы».

13.3 At-least-once в памяти и идемпотентность

Канал живёт в куче процесса. При рестарте, OOMKilled или SIGKILL содержимое исчезает целиком. Значит, in-process очередь сама по себе даёт at-most-once: задача либо обработана, либо потеряна.

Чтобы получить at-least-once, нужен источник, переживающий процесс: брокер с ack, таблица-очередь в БД или outbox. Тогда после падения задача будет доставлена снова — и появится дубль: обработчик успел сделать работу, но не успел подтвердить.

Вывод: at-least-once без идемпотентности — это генератор дублей. Ключ идемпотентности, upsert, уникальный индекс в БД или дедупликация по task_id делают повтор безопасным. В сниппете «ретрай» — это счётчик, а не доставка; в проде за ним стоит хранилище, которое переживёт рестарт.

13.4 Ретраи и отложенная очередь

Ретрай «сразу в цикле» — способ добить упавшую зависимость: сто воркеров одновременно повторяют одну и ту же ошибку. Нужны:

  • экспоненциальный backoff с потолком;
  • jitter, чтобы воркеры не синхронизировались;
  • лимит попыток и dead-letter для «ядовитых» задач;
  • классификация ошибок: 4xx и валидацию повторять бессмысленно, 5xx и таймаут — можно.

Отложенная очередь — это не time.Sleep в воркере: пока он спит, остальные задачи стоят. Варианты: min-heap по ready_at с одним таймером, time.AfterFunc, отложенные сообщения брокера.

go
delay := min(base<<attempt, maxDelay) + jitter()
time.AfterFunc(delay, func() { enqueue(task) })

13.5 Когда пора в брокер

In-process диспетчер — инструмент для работы, которую можно потерять или быстро восстановить: прогрев кэша, необязательные уведомления, метрики. Признаки, что пора выносить:

  • задача должна пережить рестарт и деплой;
  • несколько инстансов или сервисов должны её обрабатывать;
  • нужен replay, DLQ, задержки на минуты и часы;
  • важна наблюдаемость: lag, глубина очереди, возраст сообщения.

Брокер (Kafka, RabbitMQ, SQS) даёт доставку, ack, партиции и DLQ — и приносит свои проблемы: at-least-once, порядок только внутри партиции, операционную стоимость. Подробно — в mq-track.

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

  • Буфер большой vs маленький: большой сглаживает пики, но копит память и устаревшую работу; маленький честнее показывает перегрузку.
  • Блокировать vs shed: блокировка сохраняет работу ценой задержки и горутин; shed защищает сервис ценой потери задач.
  • Один воркер vs пул: один сохраняет порядок и проще; пул быстрее, но требует партиционирования по ключу.
  • Своя очередь vs брокер: своя дешевле и проще, но не даёт durability и replay; брокер надёжнее, но добавляет инфраструктуру и дубли.
  • Много попыток vs быстро в DLQ: много попыток повышает шанс доставки, но задерживает разбор ядовитой задачи.

💻 Практика

  1. Сравните политики переполнения. Диспетчер с буфером 100 и тремя режимами: блокировка, shed, отказ. Прогоните всплеск 10 000 задач при обработчике 1 мс и снимите время ответа, число принятых и потерянных задач.
  2. Сделайте порядок по ключу. Пул из 4 воркеров, но задачи одного user_id всегда обрабатывает один и тот же воркер (hash по ключу). Проверьте, что события пользователя не обгоняют друг друга.
  3. Постройте отложенные ретраи. Backoff, jitter, три попытки, затем DLQ. Добавьте «ядовитую» задачу, которая падает всегда, и убедитесь, что она не блокирует остальные.
  4. Запустите сниппет модуля: при буфере 4 из 20 задач примутся 4, отклонятся 16, воркер обработает 3, а одна уйдёт в счётчик ретраев. Вывод: «принято 4 (буфер 4), отклонено 16».

🧩 Quiz

1. Зачем к экспоненциальному backoff добавляют jitter?
2. Почему для bounded-очереди берут канал, а не слайс?
3. Какую гарантию даёт очередь в памяти процесса?
4. Когда in-process очередь пора менять на брокер?
5. Что делает политика shed при полной очереди?

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

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

  • Объяснить, почему очередь должна быть ограниченной
  • Выбрать политику переполнения под класс работы
  • Не блокировать отправку без ctx.Done()
  • Понимать границу at-most-once у очереди в памяти
  • Делать обработчики идемпотентными к повторам
  • Строить backoff с jitter, лимитом и DLQ
  • Узнавать момент, когда нужен брокер
  • Ответить на все вопросы Quiz

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

Очередь защищает фоновую работу, но данные всё равно уходят в базу — и там начинается своя конкурентность: параллельные запросы упираются в пул соединений, N+1 съедает round trip, а отмена должна доезжать до сервера. Дальше — как работать с БД, не превращая её в узкое место.