Streams вглубь: consumer groups

В прошлом уроке мы закончили со специальными структурами. Теперь возвращаемся к стримам из урока 17. Там группа mailer раздавала события заказов двум работникам, а XPENDING показал, что запись 3-0 висит на работнике m2 без подтверждения. Мы пообещали разобраться, что делать, если m2 упал. Сегодня выполняем обещание: учимся находить зависшие записи, передавать их живым работникам и не застревать на «ядовитой» записи, которая роняет всех подряд.

Почта и извещения

Почтальон взял утром десять писем и ушёл на маршрут. Восемь он вручил и отметил в журнале. На двух адресах никого не было, а сам почтальон к вечеру слёг с температурой. Письма лежат у него в сумке, а в журнале отделения напротив них стоит «выдано Иванову, не вручено».

Начальница отделения утром открывает журнал и видит: эти два письма висят на Иванове уже сутки. Она отдаёт их другому почтальону. Если письмо уже в пятый раз возвращается невручённым, его не носят снова, а кладут на отдельную полку «до востребования».

В стримах всё то же самое:

• журнал «выдано, не вручено» это список ожидающих группы;
• «висит сутки» это время простоя записи;
• «отдать другому» это команды XCLAIM и XAUTOCLAIM;
• полка «до востребования» это отдельный стрим для проблемных записей.

Исходная картина

Пять событий заказов, группа mailer, два работника. m1 взял две записи и подтвердил их, m2 взял две и упал:

127.0.0.1:6379> XGROUP CREATE events:orders mailer 0
OK
127.0.0.1:6379> XREADGROUP GROUP mailer m1 COUNT 2 STREAMS events:orders >
...
127.0.0.1:6379> XREADGROUP GROUP mailer m2 COUNT 2 STREAMS events:orders >
...
127.0.0.1:6379> XACK events:orders mailer 1-0 2-0
(integer) 2

Записи 1-0 … 5-0 добавлены так же, как в уроке 17, только с явными идентификаторами. Запись 5-0 ещё никому не выдана.

Кто что держит: XPENDING подробно

Короткая форма XPENDING из урока 17 даёт сводку. Если добавить диапазон идентификаторов и количество, Redis покажет каждую неподтверждённую запись:

127.0.0.1:6379> XPENDING events:orders mailer - + 10
1) 1) "3-0"
   2) "m2"
   3) (integer) 3113
   4) (integer) 1
2) 1) "4-0"
   2) "m2"
   3) (integer) 3113
   4) (integer) 1

У каждой записи четыре поля:

1. Идентификатор записи.
2. Работник, на котором она висит.
3. Время простоя в миллисекундах: сколько прошло с момента выдачи. Здесь чуть больше трёх секунд, у тебя число будет другим.
4. Счётчик доставок: сколько раз запись уже выдавали.

Можно оставить только то, что висит дольше заданного времени. Например, дольше минуты:

127.0.0.1:6379> XPENDING events:orders mailer IDLE 60000 - + 10
(empty array)

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

Работник вернулся: дочитать своё

Сначала простой случай. m2 не умер навсегда, его перезапустили. Перед тем как брать новые записи, он должен закончить старые. Для этого вместо > передают 0: «дай мои записи из списка ожидающих, начиная с самой первой»:

127.0.0.1:6379> XREADGROUP GROUP mailer m2 STREAMS events:orders 0
1) 1) "events:orders"
   2) 1) 1) "3-0"
         2) 1) "type"
            2) "created"
            3) "order"
            4) "16"
      2) 1) "4-0"
         2) 1) "type"
            2) "paid"
            3) "order"
            4) "16"

Это история работника: только его записи, только неподтверждённые. Новых записей так не получить, для них нужен >. Правильный работник при старте читает с 0, пока ответ не опустеет, и только потом переходит на >.

Работник не вернулся: XCLAIM и XAUTOCLAIM

Если m2 не вернётся, его записи надо отдать другому. XCLAIM передаёт конкретные записи. Число перед идентификатором это минимальное время простоя: Redis передаст запись, только если она висит хотя бы столько миллисекунд. Это защита от случая, когда два спасателя одновременно забирают одну запись.

127.0.0.1:6379> XCLAIM events:orders mailer m1 60000 3-0
(empty array)

Запись висит меньше минуты, поэтому Redis её не отдал. Для перебора всех зависших записей удобнее XAUTOCLAIM. Он сам проходит по списку ожидающих и забирает записи старше порога:

127.0.0.1:6379> XAUTOCLAIM events:orders mailer m1 2000 0-0 COUNT 1
1) "4-0"
2) 1) 1) "3-0"
      2) 1) "type"
         2) "created"
         3) "order"
         4) "16"
3) (empty array)
127.0.0.1:6379> XAUTOCLAIM events:orders mailer m1 2000 4-0 COUNT 1
1) "0-0"
2) 1) 1) "4-0"
      2) 1) "type"
         2) "paid"
         3) "order"
         4) "16"
3) (empty array)

Порог здесь 2 секунды, чтобы пример не ждал минуту. Ответ из трёх частей:

1. Курсор: с какого идентификатора продолжить. Мы просили по одной записи, поэтому первый вызов вернул курсор 4-0, а второй 0-0, то есть «прошли весь список».
2. Записи, которые теперь принадлежат m1. Он их обрабатывает и подтверждает через XACK, как обычно.
3. Идентификаторы записей, которые висели в списке ожидающих, но из самого стрима уже удалены. О них ниже, в разделе про обрезку.

После передачи XPENDING показывает нового владельца, обнулённое время простоя и выросший счётчик доставок:

127.0.0.1:6379> XPENDING events:orders mailer - + 10
1) 1) "3-0"
   2) "m1"
   3) (integer) 0
   4) (integer) 2
2) 1) "4-0"
   2) "m1"
   3) (integer) 0
   4) (integer) 2

Чтение своей истории через 0 тоже увеличивает счётчик. Каждая повторная выдача это ещё одна попытка.