XPENDING

XPENDING

XPENDING key group [[IDLE min-idle-time] start end count [consumer]]

Доступно с версии:

5.0.0

Временная сложность:

O(N) при N — количестве возвращаемых элементов, поэтому запрос небольшого фиксированного числа записей за вызов — O(1). O(M), где M — общее количество просматриваемых записей при использовании фильтра IDLE. Если команда возвращает только сводку, а список потребителей небольшой, то она выполняется за O(1) времени; в противном случае — дополнительно O(N) времени на итерацию по каждому потребителю.

Категории ACL:

@read, @stream, @slow,

Получение данных из потока через группу потребителей без подтверждения таких данных приводит к созданию ожидаемых записей. Это хорошо объясняется в команде XREADGROUP, а еще лучше в нашем введении в Redis Streams introduction to Redis Streams. Команда XACK немедленно удалит ожидаемую запись из списка ожидаемых записей (PEL), так как после успешной обработки сообщения группе потребителей больше не нужно отслеживать его и запоминать текущего владельца сообщения.

Команда XPENDING — интерфейс для проверки списка ожидающих сообщений, и поэтому является очень важной командой для наблюдения и понимания того, что происходит с группами потребителей потоков: какие клиенты активны, какие сообщения ожидают потребления или есть ли неактивные сообщения. Кроме того, эта команда вместе с XCLAIM используется для реализации восстановления потребителей, которые долгое время выходят из строя, и в результате некоторые сообщения не обрабатываются: другой потребитель может потребовать сообщение и продолжить. Это лучше объясняется в введении в потоки streams intro и на странице команды XCLAIM, и здесь не рассматривается.

Сводная форма XPENDING

Если XPENDING вызывается только с именем ключа и именем группы потребителей, то она просто выводит сводку об ожидающих сообщениях в заданной группе потребителей. В следующем примере мы создаем группу потребителей и немедленно создаем ожидающее сообщение, читая из группы с помощью XREADGROUP.

> XGROUP CREATE mystream group55 0-0
OK

> XREADGROUP GROUP group55 consumer-123 COUNT 1 STREAMS mystream >
1) 1) "mystream"
   2) 1) 1) 1526984818136-0
         2) 1) "duration"
            2) "1532"
            3) "event-id"
            4) "5"
            5) "user-id"
            6) "7782813"

Мы ожидаем, что список ожидающих записей для группы потребителей group55 в данный момент содержит сообщение: потребитель с именем consumer-123 получил сообщение без подтверждения его обработки. Простая форма XPENDING даст нам эту информацию:

> XPENDING mystream group55
1) (integer) 1
2) 1526984818136-0
3) 1526984818136-0
4) 1) 1) "consumer-123"
      2) "1"

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

Расширенная форма XPENDING

Сводка предоставляет хороший обзор, но иногда нас интересуют детали. Для того чтобы увидеть все ожидающие сообщения с более связанной информацией, нам нужно также передать диапазон идентификаторов, аналогично тому, как мы это делаем с XRANGE, и необязательный аргумент count, чтобы ограничить количество возвращаемых сообщений за вызов:

> XPENDING mystream group55 - + 10
1) 1) 1526984818136-0
   2) "consumer-123"
   3) (integer) 196415
   4) (integer) 1

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

1. Идентификатор сообщения.
2. Имя потребителя, который получил сообщение и еще не подтвердил его. Мы называем его текущим владельцем сообщения.
3. Количество миллисекунд, прошедших с момента последней доставки этого сообщения этому потребителю.
4. Количество раз, когда это сообщение было доставлено.

Счетчик доставок, который является четвертым элементом в массиве, увеличивается, когда какой-либо другой потребитель требует сообщение с помощью XCLAIM или когда сообщение снова доставляется через XREADGROUP, когда доступ к истории потребителя в группе потребителей (см. страницу XREADGROUP для получения дополнительной информации).

Можно передать дополнительный аргумент в команду, чтобы увидеть сообщения, имеющие конкретного владельца:

> XPENDING mystream group55 - + 10 consumer-123

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

Фильтр времени бездействия

Также можно фильтровать ожидающие записи потока по времени бездействия, заданному в миллисекундах (полезно для XCLAIM записей, которые не обрабатывались некоторое время):

> XPENDING mystream group55 IDLE 9000 - + 10
> XPENDING mystream group55 IDLE 9000 - + 10 consumer-123

В первом случае будут возвращены первые 10 (или меньше) записей PEL всей группы, которые простаивают более 9 секунд, а во втором случае — только те, что принадлежат consumer-123.

Исключающие диапазоны и итерация по PEL

Команда XPENDING позволяет итерироваться по ожидающим записям так же, как XRANGE и XREVRANGE позволяют итерироваться по записям потока. Это можно сделать, добавив к идентификатору последней прочитанной ожидающей записи символ (, обозначающий открытый (исключающий) диапазон, и передав его в последующий вызов команды.

Возврат

Массивный ответ, конкретно:

Команда возвращает данные в разных форматах в зависимости от способа вызова, как было описано на этой странице. Однако ответ всегда является массивом элементов.

История

• Начиная с версии Redis 6.2.0: добавлены опция IDLE и интервалы исключающих диапазонов.