Redis Streams: надёжные очереди событий

В прошлом уроке мы собрали надёжную очередь писем на списках и заплатили за это ручной работой: личные списки работников, смотритель для упавших, удаление через LREM. А когда событие «заказ создан» понадобилось бы ещё и аналитике, пришлось бы класть его во второй список. В Redis есть структура, которая всё это умеет сама: стрим.

Журнал на проходной

Вспомни журнал на проходной завода. Охранник записывает каждого входящего новой строкой внизу: время, фамилия, цех. Строки никто не вычёркивает. Отдел кадров раз в день читает журнал и отмечает, до какой строки дочитал. Бухгалтерия читает тот же журнал сама по себе и держит свою закладку. Если бухгалтер заболел, его записи никуда не делись: завтра он продолжит с закладки.

Список из прошлого урока похож на планку с чеками: снял чек, и его больше нет. Стрим похож на журнал: записи остаются, а у каждого читателя своя закладка.

Первые записи: XADD

Магазин пишет события заказов в стрим events:orders. Каждая запись это набор пар «поле значение», как в хеше:

127.0.0.1:6379> XADD events:orders * type created order 15 user 7
"1791621214587-0"
127.0.0.1:6379> XADD events:orders * type paid order 15 amount 1380
"1791621214588-0"

Звёздочка просит Redis самому выдать записи идентификатор. Он состоит из двух частей: время сервера в миллисекундах и порядковый номер внутри этой миллисекунды. У тебя цифры будут другими, потому что время другое. Если две записи попали в одну миллисекунду, Redis выдаст ...588-0 и ...588-1.

Идентификатор можно задать и руками. Так сделано в песочнице и в примерах ниже, чтобы вывод не зависел от часов:

127.0.0.1:6379> XADD events:orders 1-0 type created order 15 user 7
"1-0"
127.0.0.1:6379> XADD events:orders 2-0 type paid order 15 amount 1380
"2-0"
127.0.0.1:6379> XADD events:orders 3-0 type created order 16 user 9
"3-0"

Идентификаторы в стриме только растут. Записать что-то «в прошлое» нельзя:

127.0.0.1:6379> XADD events:orders 1-5 type broken
(error) ERR The ID specified in XADD is equal or smaller than the target stream top item

Поэтому в рабочем коде почти всегда пишут *.

Читаем журнал: XRANGE и XREAD

XRANGE возвращает записи в диапазоне идентификаторов. Минус и плюс значат «с самого начала» и «до самого конца»:

127.0.0.1:6379> XRANGE events:orders - +
1) 1) "1-0"
   2) 1) "type"
      2) "created"
      3) "order"
      4) "15"
      5) "user"
      6) "7"
2) 1) "2-0"
   2) 1) "type"
      2) "paid"
      3) "order"
      4) "15"
      5) "amount"
      6) "1380"
3) 1) "3-0"
   2) 1) "type"
      2) "created"
      3) "order"
      4) "16"
      5) "user"
      6) "9"

Каждая запись это пара: идентификатор и список полей. Мы прочитали стрим, а записи остались на месте. XLEN это подтвердит, он по-прежнему отвечает (integer) 3. XREVRANGE events:orders + - COUNT 1 читает с конца и вернёт последнюю запись.

XREAD читает записи, идущие после заданного идентификатора. Это и есть закладка: читатель запоминает последний обработанный идентификатор и в следующий раз просит всё, что после него:

127.0.0.1:6379> XREAD STREAMS events:orders 2-0
1) 1) "events:orders"
   2) 1) 1) "3-0"
         2) 1) "type"
            2) "created"
            3) "order"
            4) "16"
            5) "user"
            6) "9"

Ответ начинается с имени стрима: XREAD умеет читать сразу несколько стримов, как BRPOP ждал на нескольких списках.

У XREAD есть блокирующий режим, как у BRPOP. Вместо идентификатора пишут $, «только новые записи», и ждут до 10 секунд:

127.0.0.1:6379> XREAD BLOCK 10000 STREAMS events:live $

Пока во втором терминале никто не сделал XADD events:live ..., ответа нет. Как только запись появилась, она приходит сюда. Если за 10 секунд ничего не пришло, ответ (nil).