Pub/Sub: рассылка событий в реальном времени

Сценарий: сервис обновил профиль пользователя, и об этом должны узнать три инстанса с локальными кэшами, панель мониторинга и модуль уведомлений. Опрашивать всех по HTTP — писать опрос за опросом, и всё равно запаздывать. Redis решает это встроенной моделью издатель-подписчик: команда Publish кладёт сообщение в именованный канал, и сервер немедленно доставляет его копию каждому, кто в этот момент подписан командой Subscribe. Хендлер не знает ни одного адресата, адресаты не знают издателя — знание о событии живёт только в названии канала. В этом уроке мы подключаем подписку в горутине, читаем сообщения двумя способами, собираем реальный сценарий инвалидации кэша и разбираем главную ловушку модели: то, что не было услышано, не было сохранено. Место Pub/Sub среди структур было в обзоре типов данных Redis — здесь живём в нём целиком.

Каналы Redis: издатель не знает подписчиков#

Канал — это просто строка: events:orders, events:users, cache:invalidate. Издатель вызывает Publish с именем канала и сообщением; сервер находит всех подписчиков этого канала и отправляет им копию — без очередей, без подтверждений, без истории. Если подписчиков нет, сообщение исчезает мгновенно, и Publish честно вернёт ноль доставленных. Подписка бывает точной и шаблонной: Subscribe слушает конкретные имена, PSubscribe — паттерны вроде events:*, и одно сообщение падает в несколько подписок сразу. Принципиально понимать границу с очередью из урока про списки: там задача уходит одному воркеру и делится между потребителями, здесь копия уходит каждому подписчику. Разные модели для разных задач: очереди распределяют работу, Pub/Sub оповещает о факте. И ни одна из них не хранит то, что никого не застало — за хранение отвечает третий тип, потоки, им посвящён следующий урок.

Издатель и подписчик на Publish и Subscribe

func publishOrderCreated(ctx context.Context, id int64) error {
        payload := "created:" + strconv.FormatInt(id, 10)

        return rdb.Publish(ctx, "events:orders", payload).Err()
}

func startSubscriber(ctx context.Context) {
        sub := rdb.Subscribe(ctx, "events:orders")
        defer sub.Close()

        for msg := range sub.Channel() { // <-chan *redis.Msg
                fmt.Println(msg.Channel, msg.Payload)
        }
}

Subscribe переводит выделенное соединение в режим подписки и возвращает объект PubSub — его держат открытым всю жизнь процесса, а не на один запрос. Channel() отдаёт Go-канал с сообщениями, из которого удобно читать range-ом: у каждого сообщения есть поля Channel и Payload. Пинг-понг, переподключения и восстановление подписок go-redis делает сам; Close возвращает соединение в пул. Для запуска нужен локальный Go и Redis в Docker — песочница сайта сетевые соединения не открывает.

Подписка в горутине и остановка без утечек#

Подписка — долгоживущий ресурс, и место ей в горутине, которая запускается при старте сервиса рядом с HTTP-сервером. Чтение через Channel() блокируется до появления сообщения — это нормально: горутина спит, процессор не тратится. Вопрос в остановке: при выключении сервиса канал Go должен закрыться, а соединение — вернуться. Существует два пути: закрыть подписку через Close — range по Channel() завершится, и горутина выйдет; либо завершить контекст, который передан Subscribe, — go-redis закроет канал сообщений сам. Буфер канала настраивается опцией WithChannelSize: если читатель слишком медленный, а события сыплются, переполненный буфер приводит к разрыву соединения — go-redis считает, что подписчик не справляется, и честно отвалится, чтобы переподключиться заново. Второй способ чтения — ReceiveMessage: синхронное ожидание одного сообщения, полезное в утилитах и тестах, где горутина лишняя. Отдельно проговорите владение: один объект подписки — один читатель, две горутины на одном Channel() будут рвать поток пополам, потому что каждый элемент канала Go достаётся ровно одному получателю. Если работа над сообщением тяжёлая, разгружайте цикл: читатель принимает сообщение и передаёт в рабочий канал, обработчики трудятся параллельно, а подписка продолжает слушать без задержек.

ReceiveMessage: одно сообщение с таймаутом

func waitForEvent(ctx context.Context, sub *redis.PubSub) (string, error) {
        ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
        defer cancel()

        msg, err := sub.ReceiveMessage(ctx) // блокируется до события
        if err != nil {
                return "", err // в том числе таймаут контекста
        }
        return msg.Payload, nil
}

ReceiveMessage возвращает ровно одно сообщение и блокируется до его появления или ошибки. Таймаут контекста развязывает ожидание: без событий через десять секунд вернётся ошибка контекста, а не вечное зависание утилиты. Для постоянного чтения в сервисе удобнее Channel, а ReceiveMessage — инструмент для CLI-команд, скриптов и тестов, где подписка нужна на короткую жизнь.

Pub/Sub ничего не хранит: офлайн-подписчик теряет события молча

Сервис перезапустился на пять секунд, сетевые качели оборвали соединение — и все события за это время исчезли навсегда: ни алерта, ни уведомления, ни строки в логе. Типичная ошибка — строить на Pub/Sub очередь задач или доставку платежных уведомлений. Если потеря недопустима, берите потоки из следующего урока: у них есть история, подтверждения и дочитывание после офлайна. Число из Publish равное нулю — верный признак, что событие никого не застало.

Инвалидация кэша событием: сценарий, где Pub/Sub уместна#

Самый частый честный сценарий Pub/Sub в Go-сервисах — чистка локальных кэшей на нескольких инстансах. Схема: каждый инстанс держит горячий слой в памяти процесса, а общий кэш — в Redis; при записи в базу издатель публикует сообщение invalidated:user:42, все инстансы, подписанные на канал cache:invalidate, выкидывают этот ключ из локального слоя немедленно. Без событий каждый инстанс ждал бы собственного TTL — до десяти минут расхождения между репликами, и пользователь, обновивший аватар, видел бы старый на каждой второй странице. Событие дешевле даже короткого TTL и гарантированно синхронизирует инстансы. Обратите внимание на конструкцию: модель Pub/Sub применима именно потому, что потеря события не катастрофична — худший случай это устаревший локальный ключ до следующего TTL, а страховка остаётся на месте. Границу между стратегиями TTL, событием и write-through мы проводим в уроке про инвалидацию кэша.

Pub/Sub рядом с очередью и журналом#

Критерий	Pub/Sub	Список (очередь)	Streams
Кому доставляется	Каждому подписчику копия	Ровно одному воркеру	Одному потребителю группы
Хранение	Нет: услышанное ушло, остальное исчезло	Задача вынимается из списка	Журнал остаётся после чтения
Подтверждения	Нет	Нет	XAck и pending-список
Чтение истории	Невозможно	Только ещё не вынутое	С любой отметки, включая начало
Типовая роль	События и инвалидация в реальном времени	Фоновые задачи и буфер пиков	Журнал с гарантиями и дочитыванием

Пара вариаций, которые встречаются сразу после первого рабочего канала. Сообщения в Pub/Sub — строки без сжатия и без метаданных: не тащите в них объекты целиком, отправляйте идентификатор, а данные читайте из кэша или базы, иначе каждый инстанс получит тяжёлую копию каждого события. Паттерн-подписка PSubscribe экономит имена каналов: события сущностей уходят в events:orders, events:users, а аудит-модуль слушает events:* одним подписчиком. Права доступа не забывайте: любой клиент с доступом к Redis подписывается на любой канал, поэтому секретные вещи через Pub/Sub не рассылают. И держите в голове цену широковещательности: сто инстансов с подпиской — сто копий каждого сообщения по сети; это модель для оповещений, а не для передачи больших данных, где транспорт выбирают иначе. Ещё одна издержка — соединения: каждая подписка занимает отдельное соединение из пула, и сотня подписчиков в одном процессе выедет лимиты. Группируйте темы в несколько каналов и маршрутизируйте сообщения внутри процесса по имени канала, а не плодите подписку на каждый чих.

Прежде чем отлаживать подписку кодом, отладьте её руками: redis-cli умеет и то, и другое. В первом окне команда SUBSCRIBE events:orders переводит клиента в режим подписки, и терминал подтверждает её тремя строками. Во втором окне PUBLISH events:orders test-42 отправляет сообщение — и в первом окне мгновенно появляется тройка: тип message, имя канала, payload. Это пять секунд, которые заменяют половину отладочных логов: сразу видно, доходят ли события вообще, как выглядит payload и совпадают ли имена каналов — опечатка в имени канала самая частая причина «подписчик молчит». В Go-коде та же история проверяется проще: логируйте каждое принятое сообщение с именем канала, а для издателя — число из Publish. Ноль в логе издателя мгновенно объясняет тишину подписчика: событие отправилось, но никого не застало, и искать проблему в коде подписчика бессмысленно.

Проверка: сообщение в пустой комнате

Издатель отправил сообщение, когда ни одного подписчика не было. Что увидит подписчик, подключившийся через минуту?

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

Что означает число, которое возвращает rdb.Publish(ctx, channel, msg).Result()?

Сколько подписчиков услышит?

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

// горутина A подписана на "events:orders",
// горутина B подписана на "events:users",
// горутина C подписана на "events:orders"

n, _ := rdb.Publish(ctx, "events:orders", "created:42").Result()
fmt.Println(n)

Соберите подписку и публикацию

Допишите подписку на канал, отправку сообщения и синхронное чтение одного сообщения.

sub := rdb.(ctx, "events:orders") _, _ = rdb.(ctx, "events:orders", "created:42") msg, _ := sub.(ctx)

Соберите подписчика-горутину

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

Строки

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

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

Карточки: Pub/Sub

Термин

Publish

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

Определение

Кладёт сообщение в канал и возвращает число подписчиков, получивших его сейчас. Ноль — событие никого не застало и исчезло навсегда.

1 / 6

Модель издатель-подписчик встала на место: подписка в горутине живёт со стартом сервиса, события расходятся мгновенно, а границы применения очерчены честно — оповещение да, гарантия доставки нет. Дальше по курсу две соседние темы: урок про Streams добавляет к событиям журнал, идентификаторы и потребительские группы, а контекст и таймауты из урока про время ожидания Redis защищают длинные подписки от зависаний. Доставку событий до браузера поверх такой рассылки удобно строить через WebSocket — там подписка превращается в push клиентам, как показано в курсе про WebSocket на Go. Полный план — на странице курса по Redis; домашнее задание ниже соберёт три инстанса-подписчика в трёх терминалах, и одно из событий вы намеренно отправите в пустую комнату. Когда подписка заработает, добавьте к ней корректную остановку по сигналу: graceful shutdown подписчика — половина честной архитектуры событий, и проверить её стоит до продакшена, а не после первого инцидента деплоя.

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

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

Вывод

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

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

Pub/Sub рассылает копию сообщения всем подписчикам канала в момент публикации: издатель не знает адресатов, канал — просто имя.
Subscribe возвращает объект PubSub на выделенном соединении; Channel() отдаёт Go-канал для горутины, Close возвращает соединение.
ReceiveMessage читает одно сообщение синхронно с таймаутом контекста — способ для утилит и тестов, а не для сервисов.
Модель ничего не хранит: офлайн-подписчик теряет события молча, а Publish вернёт ноль доставленных как единственный признак.
Честный сценарий — инвалидация локальных кэшей нескольких инстансов: потеря события некритична, страховкой остаётся TTL.
В сообщениях только идентификаторы, не объекты: каждая копия падает в каждый инстанс, и тяжёлые данные дешевле читать по id.

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

Соберите издателя и подписчика из урока: publishOrderCreated и startSubscriber с чтением в горутине. Запустите два подписчика в отдельных терминалах локальным Go с Redis в Docker, отправьте событие — убедитесь, что оба увидели его и Publish вернул два. Затем остановите один подписчик, отправьте ещё одно событие и перезапустите его: пропущенное сообщение не должно догнать — прочувствуйте это. В конце добавьте логирование нуля из Publish и проверьте сценарий без подписчиков.

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

Pub/Sub или очередь на списках — что выбрать для задач?

Определите судьбу сообщения: если обработать должен один воркер — очередь на списке, каждый элемент уйдёт одному потребителю. Если узнать должны все инстансы — Pub/Sub, каждый получит копию. Задачи через Pub/Sub не распределяют: вместо одной обработки вы получите их по числу подписчиков.

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

Только сменой модели: потоки Redis хранят журнал после чтения, а потребительские группы с XAck помнят, что обработано. С Pub/Sub рестарт всегда означает потерю, и это нужно принимать конструктивно — либо как допустимую цену для оповещений, либо как сигнал выбрать Streams.

Почему подписчик разрывается, если я не успеваю читать сообщения?

go-redis держит буфер канала с ограниченным размером и таймаутом отправки. Когда читатель отстаёт и буфер переполняется, клиент разрывает соединение и переподключается заново — пропущенные сообщения при этом не возвращаются. Увеличьте WithChannelSize, ускорьте обработку или вынесите тяжёлую работу из цикла чтения в отдельную горутину.

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

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