Streams: журнал событий вместо очереди

Задача, которую простая очередь из урока про списки решает плохо: сервису нужен журнал того, что произошло. Заказы, платежи, клики — поток фактов, к которому хочется вернуться, дочитать то, что пропустил, и раздавать события нескольким обработчикам независимо. LPop вырывает задачу навсегда: после чтения нечего перечитать, а упавший воркер теряет работу. Streams закрывают это конструирование: XAdd дописывает запись в конец журнала, идентификатор растёт монотонно, чтение не удаляет ничего, а потребительские группы ведут учёт, кто что подтвердил. В этом уроке пишем события в поток, читаем их с отметки, распределяем между потребителями с XAck и обрезаем журнал, чтобы он не съел память. Место потоков среди остальных типов — в обзоре типов данных Redis, а сравнение с очередью и Pub/Sub сведём в таблицу в конце.

Stream: журнал событий с идентификаторами вместо очереди#

Поток — это append-only журнал: записи только дописываются в конец, и каждая получает идентификатор вида миллисекунды-порядковыйНомер, например 1717420000000-0. Идентификаторы монотонно растут, поэтому порядок событий встроен в сами имена — сортировать в Go ничего не нужно, а сравнение строк даёт правильную хронологию. Значения каждой записи — набор полей поле-значение, как в хэше из урока про HSet: несколько колонок на событие вместо одной строки. Главное отличие от списка — судьба после чтения: LPop удаляет элемент, а XRead ничего не трогает — двое читателей видят один и тот же журнал независимо. Длина потока — XLen, история читается диапазоном XRange, а обрезка задаётся при записи через MaxLen, чтобы журнал рос ровно до нужного горизонта. Плата за эти возможности — объём: журнал хранит всё, что в него вошло, и без обрезки превращается в самый дорогой ключ в базе.

XAdd: запись события с обрезкой журнала

func logOrderEvent(ctx context.Context, orderID int64, event string) (string, error) {
        return rdb.XAdd(ctx, &redis.XAddArgs{
                Stream: "events:orders",
                Values: map[string]any{
                        "event":    event,
                        "order_id": orderID,
                },
                MaxLen: 1000, // держим журнал ограниченным
                Approx: true, // подрезаем примерно, не блокируя запись
        }).Result()
}

XAdd принимает структуру XAddArgs: имя потока, карту полей события и, при желании, лимит длины. Возвращаемый идентификатор — строка вида 1717420000000-0: время в миллисекундах и порядковый номер внутри миллисекунды. Числа из карты клиент приведёт к строкам, как и в хэшах. MaxLen с Approx true просит Redis подрезать старые записи приблизительно: обрезка идёт узлами, не на каждом вызове, поэтому запись остаётся быстрой, а журнал — ограниченным.

XRead: чтение журнала с отметки#

Чтение потока держится на курсоре: вы говорите, с какого идентификатора готовы продолжать, а сервер отдаёт всё, что накопилось после него. Специальные значения упрощают старт: ноль означает с начала журнала — так работает переигрывание истории, а доллар — только новые события, пришедшие после вызова. Аргумент Count ограничивает порцию, Block превращает чтение в ожидание: воркер засыпает на стороне Redis и просыпается на новом событии, ровно как BLPop у списка. Отметку продвигает сам читатель — это смещение, только осмысленное и человеческое: 1717420000000-0 показывает точный момент, а не абстрактный offset. Для разовой инспекции истории есть XRange: диапазон идентификаторов от и до — половина отладки потоков делается им прямо в redis-cli. Курсор хранится у потребителя, а не в сервере: перезапуск воркера с сохранённой отметкой продолжает с того же места, что и делает потоки похожими на честные логи с offset.

Следим за потоком в цикле

func followStream(ctx context.Context) {
        lastID := "0" // читать журнал с самого начала

        for {
                streams, err := rdb.XRead(ctx, &redis.XReadArgs{
                        Streams: []string{"events:orders", lastID},
                        Count:   10,
                        Block:   5 * time.Second,
                }).Result()
                if errors.Is(err, redis.Nil) {
                        continue // таймаут: нового ничего, круг сначала
                }
                if err != nil {
                        log.Print(err)
                        time.Sleep(time.Second)
                        continue
                }
                for _, s := range streams {
                        for _, m := range s.Messages {
                                fmt.Println(m.ID, m.Values["event"])
                                lastID = m.ID // двигаем отметку
                        }
                }
        }
}

XRead отдаёт срез потоков, и в каждом — список сообщений с полями ID и Values. Курсор lastID движется руками читателя: после порции он указывает на последний обработанный идентификатор, и следующая порция начинается ровно за ним. redis.Nil на таймауте Block — штатный пустой такт. Для запуска нужен локальный Go и Redis в Docker — песочница сайта сетевые соединения не открывает.

Типичная ошибка: доллар вместо нуля при старте воркера#

Курсор в примере стартует с нуля, а в рабочем коде его часто меняют на доллар — мол, читать только новое. Пока воркер жив, разницы нет; после падения или деплоя появляется пропасть: всё, что записалось в поток за время простоя, доллар перескакивает молча. Для Redis события не потеряны — они лежат в журнале, — но для воркера их не существует: клиенты спрашивают, где письмо о заказе, а рассылка о самом существовании события даже не узнала. Правильный старт — с сохранённой отметки: последний обработанный идентификатор держите в Redis или базе, а при первом запуске берите ноль и дочитывайте историю. Исключение — потоки, где прошлое не нужно вовсе: телеметрия живых метрик, курс валют. Там доллар честен, но напишите это в комментарии к коду — иначе через полгода никто не вспомнит, почему так.

Потребительские группы: распределение с подтверждением#

Несколько воркеров на одном потоке усложняют картину: каждый держит свой курсор, и обработка задваивается. Потребительские группы решают это структурой: группа создаётся командой XGroupCreate, запоминает общую отметку, а потребители внутри группы делят записи — каждый идентификатор достаётся ровно одному, как в очереди. Чтение через XReadGroup с символом «больше» запрашивает только новые, никому не отданные записи. Дальше вступает главное отличие от списка: полученная запись не исчезает, а переезжает в pending-список группы, где числится за конкретным потребителем. Воркер обрабатывает событие и подтверждает его XAck — запись покидает pending. Если воркер умер до подтверждения, запись остаётся числиться за ним, и её можно вернуть: XPending показывает застрявшие, XAutoClaim переприсваивает их живым потребителям. Упавший обработчик больше не теряет работу — она числится, пока кто-то не подтвердит или не перераспределит.

Воркер группы: XReadGroup и XAck

func startConsumer(ctx context.Context) error {
        err := rdb.XGroupCreate(ctx, "events:orders", "workers", "0")
        if err != nil && !strings.Contains(err.Error(), "BUSYGROUP") {
                return err // группа уже существует — это нормально
        }

        for {
                streams, err := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{
                        Group:    "workers",
                        Consumer: "worker-1",
                        Streams:  []string{"events:orders", ">"},
                        Count:    10,
                        Block:    5 * time.Second,
                }).Result()
                if err != nil {
                        time.Sleep(time.Second)
                        continue
                }
                for _, s := range streams {
                        for _, m := range s.Messages {
                                handle(m.Values)
                                rdb.XAck(ctx, "events:orders", "workers", m.ID)
                        }
                }
        }
}

XGroupCreate при повторном запуске отвечает BUSYGROUP — это не авария, а признак, что группу создали раньше, и её фильтруем. Символ «больше» в Streams запрашивает только новые записи: уже отданные живут в pending и раздаются отдельными механизмами. XAck после успешной обработки снимает запись с учёта; пока подтверждения нет, запись числится за потребителем и доступна XPending и XAutoClaim — упавший воркер не прячет работу навсегда.

Забытый XAck: задача вечно в работе

Обратная ловушка гарантий: воркер обработал событие, но XAck потеряли при рефакторинге. Запись остаётся в pending, новые потребители её не увидят — символ «больше» отдаёт только неотданные, — а XPending растёт незаметно. Типичный симптом: события «выполняются», но результат нигде не виден. Держите метрику длины pending и ревизию XAutoClaim — она возвращает застрявшее живым воркерам.

Кейс: два читателя вместо группы — письмо ушло дважды#

Инцидент, который всплывает на разборах чужих систем до сих пор: сервис уведомлений и сервис аналитики читали один поток заказов через отдельные XRead с собственными курсорами. Каждому казалось, что они просто независимы, — и оба сервиса получали каждую запись. Аналитика считала себе тихо, а уведомления задваивались: клиент получал по два письма о каждой оплате, и поддержку неделю водило кругами вокруг почтового провайдера. Группы существуют ровно для этого случая: события, которые должны быть обработаны один раз, читаются через XReadGroup в общую группу — идентификатор достаётся ровно одному потребителю. Раздельные курсоры оставьте задачам, которым повтор не страшен: пересчёт отчётов, индексация, прогрев кэша. Правило на стену: обработать один раз — группа, прочитать каждому — отдельные XRead, и смешивать их в одном потоке нельзя без осознанных дублей.

Список, Pub/Sub, Streams: сравнение на одной странице#

Критерий	Список	Pub/Sub	Streams
Доставка	Одному воркеру, без подтверждения	Всем подписчикам копией	Одному потребителю группы, с подтверждением
После чтения	Задача удалена навсегда	Ничего не хранится	Запись остаётся в журнале
Падение воркера	Задача потеряна	События пропущены молча	Запись числится в pending и перераспределяется
История	Только не вынутое	Отсутствует	Читается с любой отметки
Сложность	Две команды	Две команды	Группы, ack и pending — дороже в обслуживании

Как выбрать из трёх моделей, не переусложнив: если нужен конвейер задач и потеря при падении приемлема или перекрыта RPopLPush — списка хватает, его две команды проще потока с группами. Если события должны дойти до всех инстансов немедленно и потеря не страшна — Pub/Sub из прошлого урока короче и дешевле. Журнал с гарантиями — территория Streams: платежи, аудит, события, которые обязаны быть обработаны хотя бы раз и дочитаны после сбоя. Внутри потока пригодятся те же привычки, что и везде в курсе: обрезка MaxLen вместо вечного роста, идентификаторы в сообщениях вместо объектов целиком, имена потоков по сущностям — events:orders, events:payments. Воркеры на горутинах из курса по продвинутому Go ложатся поверх XReadGroup один к одному: один читатель, пул обработчиков, XAck по результату.

Состояние группы — два запроса XINFO

Когда дежурный спрашивает, жива ли группа, отвечают XINFO STREAM events:orders и XINFO GROUPS events:orders: длина журнала, отметка группы, число потребителей и размер pending. В go-redis это методы XInfoStream и XInfoGroups — повесьте их дамп в метрики раз в минуту, и застрявшая группа станет видна на графике раньше первого тикета.

Проверка: что возвращает XAdd

Что вернёт rdb.XAdd(ctx, &redis.XAddArgs{Stream: "events:orders", Values: ...}).Result()?

Проверка: воркер умер до XAck

Потребитель получил запись через XReadGroup, обработка упала, XAck не вызван. Что станет с записью?

Правда ли, что вторая запись позже?

Прочитайте код и выберите, что он выведет:

id1, _ := rdb.XAdd(ctx, &redis.XAddArgs{
        Stream: "events:orders",
        Values: map[string]any{"event": "created"},
}).Result()

id2, _ := rdb.XAdd(ctx, &redis.XAddArgs{
        Stream: "events:orders",
        Values: map[string]any{"event": "paid"},
}).Result()

fmt.Println(id1 < id2)

Соберите запись в поток

Допишите команду записи, поле с данными события и режим приблизительной обрезки журнала.

id, _ := rdb.(ctx, &redis.XAddArgs{ Stream: "events:orders", : map[string]any{"event": "created", "order_id": 42}, MaxLen: 1000, : true, })

Соберите цикл воркера группы

Расставьте строки по порядку: чтение новой порции записей группой, обработка и подтверждение каждой.

Строки

Ваша программа

Нажимайте на строки слева, чтобы собрать программу

Карточки: Streams

Термин

Stream

нажмите, чтобы перевернуть

Определение

Append-only журнал событий: записи дописываются в конец, каждая получает монотонный идентификатор вида миллисекунды-порядковый. Чтение ничего не удаляет.

1 / 6

Потоки закрыли линейку структур данных курса: строки, хэши, списки, множества, ZSET и теперь журнал с гарантиями — тип подобран под задачу, а не наоборот. Следующие уроки уходят в инженерию клиента: пайплайны из урока про один рейс сети ускорят массовые записи, а транзакции из урока про MULTI и WATCH склеят чтение и запись атомарно — тот случай, когда XAdd и ответная запись обязаны случиться вместе. Тесты для воркеров потоков удобно гонять на miniredis — фейковый Redis в памяти разбирался в уроке про тесты без Docker. Полный план курса — на странице курса по Redis; домашнее задание попросит сломать XAck и найти застрявшую запись через XPending — лучший способ запомнить pending — увидеть его своими глазами в redis-cli.

Песочница урока

Редактируйте код и запускайте его прямо здесь — интерпретатор работает в браузере, ничего устанавливать не нужно. За каждый запуск — +2 XP.

Вывод

Запустите код, чтобы увидеть вывод.

Коротко: что запомнить

Stream — append-only журнал: XAdd дописывает запись с монотонным идентификатором, чтение ничего не удаляет.
XRead читает с курсора: ноль — с начала, доллар — только новое; Block превращает чтение в ожидание, отметку двигает читатель.
Потребительские группы делят записи между воркерами: каждый идентификатор достаётся ровно одному, как в очереди.
Запись живёт в pending до XAck: XPending и XAutoClaim возвращают застрявшее — падение воркера не теряет работу.
MaxLen с Approx держит журнал ограниченным, подрезая старые записи без блокировки записи.
Выбор модели: список — простая очередь, Pub/Sub — оповещения, Streams — события с гарантиями и историей.

Домашнее задание

Соберите logOrderEvent, followStream и startConsumer из урока, запустите локальным Go с Redis в Docker. Напишите пять событий, убедитесь, что followStream переиграл их с нуля, а воркер группы обработал и подтвердил каждую. Затем сломайте XAck: закомментируйте подтверждение, обработайте порцию и посмотрите XPENDING events:orders workers в redis-cli — записи останутся в pending. Верните XAck, а после этого добавьте ревизию XAutoClaim и добейтесь, чтобы застрявшее возвращалось живому воркеру.

Вопросы и ответы по уроку

Когда Streams избыточны?

Когда потеря задачи приемлема или событие должно дойти до всех: список проще для конвейера, Pub/Sub — для оповещений. Поток приносит группы, ack и pending — плюс три понятия к коду и эксплуатации. Если вам нужны только две команды и скорость, а гарантии перекрываются повторами на стороне отправителя, поток окупится не сразу.

Какие сообщения хранить в потоке?

Идентификаторы и факты: идентификатор заказа, тип события, отметку времени. Тяжёлые объекты читайте по идентификатору из базы или кэша — журнал копирует данные каждому, кто его перечитывает, и объём растёт быстрее, чем кажется. Обрезка MaxLen задаётся на запись, потому что после переполнения чистить журнал руками уже поздно.

Сколько потребителей можно держать в группе?

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

Термины урока

Ключевые понятия из этого урока — короткие определения в глоссарии. Наведите на термин, чтобы вспомнить, откройте карточку для подробностей и ссылки на практику.