Consumer groups и надёжные очереди

Consumer groups превращают поток в полноценную очередь с балансировкой воркеров, подтверждением обработки и ретраями. Это сердце паттерна «очередь задач на Redis Streams».

Модель группы

• Группа прикреплена к потоку и имеет последний доставленный ID;
• В группе N потребителей (имена придумываете вы); каждая запись доставляется ровно одному свободному потребителю — балансировка автоматическая;
• Взятая запись попадает в PEL (pending entries list) группы и остаётся там, пока не придёт XACK;
• Разные группы независимы: каждая получает все записи потока.

Жизненный цикл сообщения

# 1. Создать группу ($ — с новых записей; 0 — читать поток с начала)
XGROUP CREATE orders workers $

# 2. Воркер забирает записи: ">" = только ещё не доставленные группе
XREADGROUP GROUP workers worker-1 COUNT 10 BLOCK 5000 STREAMS orders >

# 3. Успешная обработка → подтверждение
XACK orders workers 1717000000000-0

# 4. Что "висит" незавершённым в PEL
XPENDING orders workers - + 10          # сводка
count: 2  consumer: worker-1  idle: 120000  delivery-count: 1 ...

# 5. Воркер умер — другой перехватывает записи, висящие > 60с
XAUTOCLAIM orders workers worker-2 60000 10
next-cursor, claims[{id, fields}], deleted-ids

# 6. После N попыток доставки (delivery-count) — в dead-letter
XADD orders:dlq * orig_id 1717000000000-0 payload ...
XACK orders workers <orig_id> ; XDEL orders <orig_id>

Семантика: at-least-once

Перехват чужих зависших записей означает: обработка может повториться. Поэтому:

• обработчик обязан быть идемпотентным. Простой приём — дедупликация по ID:

  # Успех дедупликации = SET вернул OK (ключ был новый)
  SET processed:{msg_id} 1 NX EX 86400

• idle-таймаут перехвата должен быть больше времени обработки, иначе живого воркера «подрежут» конкурентно;
• XDEL убирает запись из потока, но не из PEL — потому XAUTOCLAIM возвращает и список deleted-ids: их надо ACK'нуть.

Шаблон «надёжная очередь задач»

1. Продюсер: XADD tasks MAXLEN ~ 100000 * type ... payload ...;
2. Пул воркеров: XREADGROUP GROUP g w1 COUNT 10 BLOCK 5000 STREAMS tasks >;
3. Идемпотентная обработка + XACK;
4. Отдельный «пастух»: раз в 10с XAUTOCLAIM (idle > обработки × 2), delivery-count > N → XADD в tasks:dlq + XACK;
5. Retention: XTRIM периодически; для задач — MINID по времени;
6. Мониторинг: XLEN (рост = не успеваем), XPENDING (висяки), длина DLQ (алерт).

Streams vs Lists для очередей

List до сих пор отлично работает для простых fire-and-forget задач. Как только появляется «надо гарантировать обработку» — переходите на Streams.

Сравнение с Kafka и RabbitMQ

Практика

1. Прогоните весь жизненный цикл из урока в двух терминалах: XADD → XREADGROUP → (убейте воркера до XACK) → XPENDING → XAUTOCLAIM другим воркером → XACK.
2. Реализуйте DLQ-логику с delivery-count и проверьте: задача, падающая 4 раза, должна оказаться в orders:dlq.
3. Замерьте пропускную способность: redis-benchmark -t xadd -n 100000 -d 64 против lpush — оцените порядок цифр на вашем железе.