Pub/Sub против Streams: что выбрать

В прошлом уроке мы выжимали из Redis скорость. Теперь вернёмся к сообщениям между сервисами. У Redis для них три инструмента: Pub/Sub, очереди на списках и стримы с группами потребителей. Каждый мы уже пробовали по отдельности. Сегодня ставим их рядом и учимся выбирать под задачу.

Радио, почтовый ящик и журнал

Магазину нужно сообщить, что заказ 15 оплачен. Сообщить можно тремя способами.

• Радио. Диктор говорит в эфир. Кто включил приёмник, тот услышал. Кто не включил, не узнает никогда: записи эфира нет. Это Pub/Sub.
• Почтовый ящик. Письмо лежит в ящике, пока его не заберут. Но забирает его кто-то один: второй сотрудник найдёт пустой ящик. Это список.
• Журнал на проходной. Каждая запись остаётся в журнале. У бухгалтерии своя закладка, у склада своя, и каждый отдел прочитает все записи в своём темпе. Это стрим.

Ни один способ не лучше остальных. Радио самое лёгкое, журнал самый надёжный, а ящик посередине.

Одно событие тремя способами

Отправим событие «заказ 15 оплачен» всеми тремя способами сразу, а получателей подключим позже:

127.0.0.1:6379> PUBLISH orders:paid order:15
(integer) 0
127.0.0.1:6379> LPUSH queue:emails order:15
(integer) 1
127.0.0.1:6379> XADD events:orders 1-0 type paid order 15
"1-0"

PUBLISH ответил 0: слушателей не было, сообщение пропало. Теперь приходят опоздавшие получатели. Два работника по очереди читают список:

127.0.0.1:6379> RPOP queue:emails
"order:15"
127.0.0.1:6379> RPOP queue:emails
(nil)

Первый получил письмо, второй нашёл пустой ящик. А в стрим заглядывают две группы, рассылка и аналитика:

127.0.0.1:6379> XGROUP CREATE events:orders analytics 0
OK
127.0.0.1:6379> XGROUP CREATE events:orders mailer 0
OK
127.0.0.1:6379> XREADGROUP GROUP mailer m1 COUNT 10 STREAMS events:orders >
1) 1) "events:orders"
   2) 1) 1) "1-0"
         2) 1) "type"
            2) "paid"
            3) "order"
            4) "15"
127.0.0.1:6379> XREADGROUP GROUP analytics a1 COUNT 10 STREAMS events:orders >
1) 1) "events:orders"
   2) 1) 1) "1-0"
         2) 1) "type"
            2) "paid"
            3) "order"
            4) "15"

Обе группы получили событие, хотя подключились после отправки. В этом весь выбор: кто должен получить сообщение и что делать, если получателя в момент отправки нет.

Четыре вопроса к задаче

Когда в проекте появляется новое событие, задай про него четыре вопроса.

1. Можно ли потерять сообщение?

Здесь пригодятся два названия для гарантий доставки. «Не больше одного раза» значит, что сообщение может потеряться, но дублей не будет. Так работает Pub/Sub и список с простым RPOP: работник забрал задачу и упал, задачи больше нигде нет.

«Хотя бы один раз» значит, что сообщение не потеряется, но после сбоя может прийти повторно. Так работает список с LMOVE и полочкой «в работе» и стрим с XACK. Работник обработал запись, но упал до подтверждения. Запись осталась в списке ожидающих, и её выдадут снова.

Раз дубли возможны, обработчик должен быть идемпотентным: повторная обработка не должна ничего ломать. Например, перед отправкой письма проверять, не отправлено ли оно уже: SET sent:order:15 1 NX EX 86400 вернёт (nil), если письмо уже ушло.

2. Нужна ли история?

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

3. Сколько получателей?

• Pub/Sub отдаёт каждое сообщение всем подписчикам сразу. Три сервера сайта подписаны на prices:changed, и все три сбросят свой кэш.
• Список отдаёт каждую задачу одному работнику. Это правильно для задач вроде «отправить письмо»: письмо должно уйти один раз.
• Стрим умеет и то и другое. Внутри группы записи делятся между работниками, как в списке. Разные группы получают все записи, как подписчики в Pub/Sub.

4. Что, если получатель не успевает?

Сообщения приходят быстрее, чем их обрабатывают. Такую ситуацию называют обратным давлением, и каждый инструмент ведёт себя по-своему.

Pub/Sub копит непрочитанные сообщения в памяти сервера, в буфере подписчика. У буфера есть предел:

127.0.0.1:6379> CONFIG GET client-output-buffer-limit
1) "client-output-buffer-limit"
2) "normal 0 0 0 slave 268435456 67108864 60 pubsub 33554432 8388608 60"

Часть pubsub 33554432 8388608 60 читается так: если буфер подписчика вырос до 32 МБ или держится больше 8 МБ дольше 60 секунд, Redis разрывает соединение. Медленный подписчик не уронит сервер, но потеряет все накопленные сообщения.

Список и стрим копят задачи в памяти Redis, пока её хватает. Дальше вступают правила maxmemory из урока про вытеснение. Стрим обычно обрезают при записи: XADD events:orders MAXLEN ~ 100000 * ... держит около ста тысяч последних записей. Отставание группы показывает XINFO GROUPS в поле lag: сколько записей группа ещё не прочитала. Если lag растёт, работников пора добавлять.