Модуль 1. Redis как брокер: место в ландшафте, Pub/Sub и Streams

Цель модуля: понять, почему Redis вообще оказывается брокером и чего это стоит; различать Pub/Sub и Streams и выбирать между ними осознанно; уверенно работать с consumer groups Streams (XREADGROUP, pending, XACK, claim) и знать, где проходят границы применимости, за которыми нужен Kafka или RabbitMQ.

Перед началом: Модуль 1 курса «Фундамент» (гарантии доставки, идемпотентность), Docker, базовый Redis (SET/GET — не обязательно). Инструменты: redis:7, redis/redis-stack (Insight UI), github.com/redis/go-redis/v9. Трудоёмкость: ~8–10 ч (теория 4 ч · практика 4 ч · квиз 1 ч).

📖 1.1. Почему Redis оказывается брокером

Redis почти всегда уже есть в проекте: кэш, распределённые замки, rate limiter, счётчики. Когда появляется задача «нужна очередь событий», звучит знакомое: «у нас же уже стоит Redis — зачем ещё один брокер в инфраструктуре?»

Это не глупость. У «очереди в Redis» есть настоящие плюсы:

• Ноль новой инфраструктуры — тот же сервер, тот же клиент, никакого дежурного обучения;
• Микросекундные задержки — Redis-операции в памяти быстрее, чем публикация в Kafka на порядок;
• Один инструмент для смежных задач — идемпотентность через SETNX и очередь через Streams живут в одной системе.

И настоящая цена, которую платят при инцидентах:

• Память. Streams хранится в RAM (плюс disk-backed до определённой отметки). Retention измеряется сотнями мегабайт, а не гигабайтами дискового лога, как в Kafka;
• Персистентность best-effort. AOF everysec допускает потерю до секунды записей при падении процесса; вопрос «а данные точно не пропадут?» требует честного ответа;
• Очередь конкурирует за ресурс с кэшем. Раздувшийся стрим вытесняет горячие данные — или попадает под eviction сам.

Правило этого курса: Redis как брокер — осознанное решение под конкретные ограничения (малые объёмы, низкая задержка важнее ретеншена, потеря секунды допустима), а не экономия на отдельном брокере. Дальше — инструменты, чтобы это решение принимать на цифрах.

📖 1.2. Pub/Sub: вещание без памяти

Простейший механизм — каналы Pub/Sub:

go
rdb := redis.NewClient(&redis.Options{Addr: "localhost:6379"})

// Издатель
err := rdb.Publish(ctx, "prices", `{"symbol":"BTC","price":68000}`).Err()

// Подписчик
sub := rdb.Subscribe(ctx, "prices")
defer sub.Close()
ch := sub.Channel()
for msg := range ch {
    fmt.Println(msg.Channel, msg.Payload)
}

У модели «вещание» есть цена — доставка at-most-once:

• сообщение уходит только подписчикам, подключённым в момент публикации;
• нет подтверждений, нет повторов, нет истории — подписчик не в сети, значит сообщения не существует;
• буфер канала подписчика ограничен (Client-output-buffer-limit pubsub); медленный подписчик молча вылетает в разрыв соединения.

Когда Pub/Sub — правильный выбор: инвалидация кэшей, сигналинг между инстансами, live-обновления, где пропуск обновления допустим (следующее догонит). Когда нет: всё, где сообщение олицетворяет бизнес-факт — «заказ оплачен» терять нельзя.

Сравнение с семантиками из ядра трека: Pub/Sub — это даже слабее fire-and-forget: у сообщения нет получателя, а не только подтверждения.

📖 1.3. Streams: лог в Redis

Streams — ответ Redis на вопрос «а если нужно помнить?». Это append-only лог, устроенный как Kafka в миниатюре: записи с ID, чтение с позиции, группы потребителей.

go
// Продюсер: добавить событие в стрим
id, err := rdb.XAdd(ctx, &redis.XAddArgs{
    Stream: "orders",
    // Примерный лимит длины: старые записи обрезаются (~ = быстрее, чем точный)
    MaxLen: 100_000,
    Approx: true,
    Values: map[string]interface{}{
        "type":     "order.paid",
        "order_id": "o-42",
        "amount":   "1234.50",
    },
}).Result()
fmt.Println("добавлено с id:", id) // 1712345678901-0 — миллисекунды-счётчик

Что важно в модели:

• ID = время-счётчик (<unix_ms>-<seq>) — даёт порядок и «прочитай, что было после момента X» без оффсетов;
• Чтение с позиции: XREAD от $ (только новые) или от конкретного ID (догнать историю) — контраст с Pub/Sub, где истории нет;
• MaxLen/Approx — ретеншен задаётся руками: 100_000 записей или ~-аппроксимация (обрезает по целым нодам — быстрее, лимит «примерный»);
• Размер записи — ваша ответственность: большие тела (мегабайты) в стриме — прямой путь к раздуванию памяти и медленным чтениям. Кладите payload в S3/БД, а в стрим — ссылку и метаданные.

📖 1.4. Consumer groups: надёжный воркер в Redis

Настоящий брокерский режим включается с consumer groups: несколько консьюмеров делят записи, каждое подтверждение отслеживается.

go
// Один раз: создать группу (0 — читать стрим с начала)
err := rdb.XGroupCreateMkStream(ctx, "orders", "billing", "0").Err()

// Консьюмер billing-1: забираем до 10 новых записей и «держим» их,
// пока не подтвердим
streams, err := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{
    Group:    "billing",
    Consumer: "billing-1",
    Streams:  []string{"orders", ">"}, // ">" — только новые, ещё не доставленные
    Count:    10,
    Block:    2 * time.Second,
}).Result()

for _, s := range streams {
    for _, msg := range s.Messages {
        if err := handle(msg); err != nil {
            continue // не подтверждаем: запись останется в pending
        }
        // Подтверждаем только после успешной обработки
        rdb.XAck(ctx, "orders", "billing", msg.ID)
    }
}

Ключевые следствия механики (это и есть «брокер внутри Redis»):

• At-least-once по умолчанию. Не отправил XACK — запись остаётся в pending (PEL, pending entries list) у этого консьюмера и может быть переобработана. Ваш обработчик обязан быть идемпотентным — как в ядре трека;
• Упал консьюмер с pending-записями? Их надо спасать: XPENDING показывает сводку, XCLAIM (или XAUTOCLAIM — пакетная версия, Redis 6.2+) передаёт записи живому консьюмеру:

go
// Забрать «зависшие» записи старше 30 секунд молчания
claimed, _, err := rdb.XAutoClaim(ctx, &redis.XAutoClaimArgs{
    Stream:   "orders",
    Group:    "billing",
    Consumer: "billing-2",
    MinIdle:  30 * time.Second,
    Start:    "0",
    Count:    10,
}).Result()

• XACK не удаляет запись из стрима — он лишь снимает её с учёта группы. Хранение по-прежнему ограничивается MaxLen. Урок: подтверждение и удаление — разные вещи, в Kafka — та же пара (commit vs retention).

Сравните с ядром трека: prefetch в RabbitMQ = Count в XREADGROUP; XACK = ack; PEL = сообщения «в работе»; XAUTOCLAIM = redelivery при падении консьюмера. Redis повторяет знакомый контур — вопрос лишь в цене каждого звена.

📖 1.5. Гарантии и персистентность: что переживёт рестарт

До сих пор мы смотрели на механику. Теперь — на физику: Redis живёт в памяти, и честный ответ «что случится с моим стримом при отказе» зависит от настроек персистентности:

Дальше — репликация: реплика асинхронная (подтверждение клиенту — до записи на реплику), значит при падении мастера возможна потеря хвоста, даже с AOF. Sentinel переключает мастерство автоматически, но не отменяет это окно; Cluster шардирует стримы по слотам — consumer group работает, но «глобального порядка» между слотами нет.

Практический вывод для постановки задачи: Redis-очередь годится там, где потеря секунды и повторная обработка приемлемы, а пропуск — нет. Для денег, аудита и событийных логов берите систему, для которой персистентность — контракт, а не настройка. Точные сценарии отказов разберём в модулях про Sentinel/Cluster и инциденты.

💻 Практика

Лабораторная «Streams-воркер с claim'ом» — соберите локально:

1. Инфраструктура:

   bash
   docker run -d --name redis-course -p 6379:6379 redis:7
   # вкл. AOF everysec — в командной строке запуска:
   #   redis-server --appendonly yes --appendfsync everysec

2. Продюсер: раз в 100 мс пишет событие orders (XADD, MaxLen≈1000, Approx) со счётчиком в значений.
3. Два консьюмера billing-1 и billing-2 в одной группе: XReadGroup c Block: 2s; обработка «случайно» падает на каждом 5-м сообщении — без XACK (симулируйте return до подтверждения).
4. Спасатель: третья горутина раз в 5 секунд делает XAutoClaim (MinIdle 3s) и дообрабатывает зависшие — с журналом «кто чей долг забрал».
5. Проверки: XPENDING orders billing — пустой список после разбора; сообщений «без капризов» не потеряно; на «проблемных» в журнале видна пара «упал → переобработан».
6. Эксперимент с персистентностью: остановите Redis без AOF и с ним (appendonly yes) — сравните, что уцелело. Это к §1.5.

Бонус: замерьте XADD p99 latency при MaxLen = 0 (без лимита) и при MaxLen ≈ 10 000, Approx — почувствуйте цену обрезки.

🧩 Quiz "Проверь себя"

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

🔗 Мостик: дальше в курсе

Мы взяли базовый контур: Pub/Sub, Streams, группы и подтверждения. Дальше — эксплуатационная глубина: Sentinel и Cluster (что именно теряется при failover и как это выглядит в цифрах), мониторинг (latency, фрагментация памяти, slowlog, длина стримов и pending-лаг — что алертить), инциденты (big/hot keys, eviction под нагрузкой, OOM). И честное сравнение «Streams vs Kafka vs RabbitMQ» — таблицей, чтобы защищать решение на ревью.