Redis Stream

Redis Stream — это структура данных, недавно добавленная в Redis версии 5.0.

Redis Stream в основном используется для очередей сообщений (MQ, Message Queue). В самом Redis есть функция публикации/подписки (pub/sub) для реализации функциональности очереди сообщений, но у неё есть недостаток: сообщения не могут быть сохранены. Если произойдёт разрыв сети, сбой Redis и т. п., сообщения будут отброшены.

Проще говоря, публикация/подписка (pub/sub) может распространять сообщения, но не может записывать исторические сообщения.

Redis Stream обеспечивает персистентность сообщений и репликацию master-slave, позволяя любому клиенту получать доступ к данным в любое время, а также запоминает позицию доступа каждого клиента и гарантирует, что сообщения не будут потеряны.

Структура Redis Stream показана ниже. Она имеет связный список сообщений, который связывает все добавленные сообщения воедино. Каждое сообщение имеет уникальный ID и соответствующее содержимое:

Каждый Stream имеет уникальное имя, которое является ключом Redis. Он создаётся автоматически, когда мы впервые используем команду xadd для добавления сообщения.

Пояснение к диаграмме:

  • Группа потребителейГруппа потребителей, создаваемая с помощью команды XGROUP CREATE. В группе потребителей может быть несколько потребителей.
  • last_delivered_idКурсор. У каждой группы потребителей есть курсор last_delivered_id. Когда любой потребитель читает сообщение, курсор last_delivered_id продвигается вперёд.
  • pending_idsПеременная состояния потребителя, используемая для хранения неподтверждённых идентификаторов потребителя. pending_ids хранит сообщения, которые уже были прочитаны клиентом, но ещё не были подтверждены (acked, Acknowledge character: символ подтверждения).

Команды, связанные с очередями сообщений:

  • XADD- Добавить сообщение в конец
  • XTRIM- Обрезать поток, чтобы ограничить его длину
  • XDEL- Удалить сообщение
  • XLEN- Получить количество элементов, содержащихся в потоке, т. е. длину сообщения
  • XRANGE- Получить список сообщений, который автоматически отфильтровывает удалённые сообщения
  • XREVRANGE- Получить список сообщений в обратном порядке, ID от большего к меньшему
  • XREAD- Получить список сообщений в блокирующем или неблокирующем режиме

Команды, связанные с группами потребителей:

  • XGROUP CREATE- Создать группу потребителей
  • XREADGROUP GROUP- Читать сообщения из группы потребителей
  • XACK- Пометить сообщение как "обработанное"
  • XGROUP SETID- Установить новый ID последнего доставленного сообщения для группы потребителей
  • XGROUP DELCONSUMER- Удалить потребителя
  • XGROUP DESTROY- Удалить группу потребителей
  • XPENDING- Отобразить информацию о сообщениях, ожидающих обработки
  • XCLAIM- Передать владение сообщением
  • XINFO - Просмотреть информацию о потоках и группах потребителей;
  • XINFO GROUPS- Вывести информацию о группах потребителей;
  • XINFO STREAM- Вывести информацию о потоке

XADD

Используйте XADD для добавления сообщения в очередь. Если указанная очередь не существует, очередь будет создана. Формат синтаксиса XADD:

XADD key ID field value [field value ...]
  • keyИмя очереди; создайте её, если она не существует
  • IDИдентификатор сообщения. Мы используем * чтобы указать, что он генерируется Redis. Его можно задать самостоятельно, но необходимо обеспечить его возрастание.
  • field valueЗапись.

Пример

redis> XADD mystream * name Sara surname OConnor
"1601372323627-0"
redis> XADD mystream * field1 value1 field2 value2 field3 value3
"1601372323627-1"
redis> XLEN mystream
(целое число) 2
redis> XRANGE mystream - +
1) 1) "1601372323627-0"
   2) 1) "name"
      2) "Sara"
      3) "surname"
      4) "OConnor"
2) 1) "1601372323627-1"
   2) 1) "field1"
      2) "value1"
      3) "field2"
      4) "value2"
      5) "field3"
      6) "value3"
redis>

XTRIM

Используйте XTRIM, чтобы обрезать поток и ограничить его длину. Формат синтаксиса:

XTRIM key MAXLEN [~] count
  • keyИмя очереди
  • MAXLENДлина
  • countКоличество

Пример

127.0.0.1:6379> XADD mystream * field1 A field2 B field3 C field4 D
"1601372434568-0"
127.0.0.1:6379> XTRIM mystream MAXLEN 2
(целое число) 0
127.0.0.1:6379> XRANGE mystream - +
1) 1) "1601372434568-0"
   2) 1) "field1"
      2) "A"
      3) "field2"
      4) "B"
      5) "field3"
      6) "C"
      7) "field4"
      8) "D"
127.0.0.1:6379>

redis>

XDEL

Используйте XDEL для удаления сообщений. Формат синтаксиса:

XDEL key ID [ID ...]
  • keyИмя очереди
  • IDID сообщения

Пример

> XADD mystream * a 1
1538561698944-0
> XADD mystream * b 2
1538561700640-0
> XADD mystream * c 3
1538561701744-0
> XDEL mystream 1538561700640-0
(целое число) 1
127.0.0.1:6379> XRANGE mystream - +
1) 1) 1538561698944-0
   2) 1) "a"
      2) "1"
2) 1) 1538561701744-0
   2) 1) "c"
      2) "3"

XLEN

Используйте XLEN, чтобы получить количество элементов, содержащихся в потоке, то есть длину сообщения. Формат синтаксиса:

XLEN key
  • keyИмя очереди

Пример

redis> XADD mystream * item 1
"1601372563177-0"
redis> XADD mystream * item 2
"1601372563178-0"
redis> XADD mystream * item 3
"1601372563178-1"
redis> XLEN mystream
(целое число) 3
redis>

XRANGE

Используйте XRANGE, чтобы получить список сообщений, который автоматически отфильтровывает удалённые сообщения. Формат синтаксиса:

XRANGE key start end [COUNT count]
  • keyИмя очереди
  • startНачальное значение,-представляет минимальное значение
  • endКонечное значение,+представляет максимальное значение
  • countКоличество

Пример

redis> XADD writers * name Virginia surname Woolf
"1601372577811-0"
redis> XADD writers * name Jane surname Austen
"1601372577811-1"
redis> XADD writers * name Toni surname Morrison
"1601372577811-2"
redis> XADD writers * name Agatha surname Christie
"1601372577812-0"
redis> XADD writers * name Ngozi surname Adichie
"1601372577812-1"
redis> XLEN writers
(целое число) 5
redis> XRANGE writers - + COUNT 2
1) 1) "1601372577811-0"
   2) 1) "name"
      2) "Virginia"
      3) "surname"
      4) "Woolf"
2) 1) "1601372577811-1"
   2) 1) "name"
      2) "Jane"
      3) "surname"
      4) "Austen"
redis>

XREVRANGE

Используйте XREVRANGE, чтобы получить список сообщений, который автоматически отфильтровывает удалённые сообщения. Формат синтаксиса:

XREVRANGE key end start [COUNT count]
  • keyИмя очереди
  • endКонечное значение,+представляет максимальное значение
  • startНачальное значение,-представляет минимальное значение
  • countКоличество

Пример

redis> XADD writers * name Virginia surname Woolf
"1601372731458-0"
redis> XADD writers * name Jane surname Austen
"1601372731459-0"
redis> XADD writers * name Toni surname Morrison
"1601372731459-1"
redis> XADD writers * name Agatha surname Christie
"1601372731459-2"
redis> XADD writers * name Ngozi surname Adichie
"1601372731459-3"
redis> XLEN writers
(целое число) 5
redis> XREVRANGE writers + - COUNT 1
1) 1) "1601372731459-3"
   2) 1) "name"
      2) "Нгози"
      3) "surname"
      4) "Адичи"
redis>

XREAD

Используйте XREAD, чтобы получить список сообщений в блокирующем или неблокирующем режиме. Формат синтаксиса:

XREAD [COUNT count] [BLOCK milliseconds] STREAMS key [key ...] id [id ...]
  • countКоличество
  • millisecondsНеобязательно. Миллисекунды блокировки. Если не задано, используется неблокирующий режим.
  • keyИмя очереди
  • idИдентификатор сообщения

Пример

# Прочитать два сообщения из начала Stream
> XREAD COUNT 2 STREAMS mystream writers 0-0 0-0
1) 1) "mystream"
   2) 1) 1) 1526984818136-0
         2) 1) "duration"
            2) "1532"
            3) "event-id"
            4) "5"
            5) "user-id"
            6) "7782813"
      2) 1) 1526999352406-0
         2) 1) "duration"
            2) "812"
            3) "event-id"
            4) "9"
            5) "user-id"
            6) "388234"
2) 1) "writers"
   2) 1) 1) 1526985676425-0
         2) 1) "name"
            2) "Virginia"
            3) "surname"
            4) "Woolf"
      2) 1) 1526985685298-0
         2) 1) "name"
            2) "Jane"
            3) "surname"
            4) "Austen"

XGROUP CREATE

Используйте XGROUP CREATE для создания группы потребителей. Формат синтаксиса:

XGROUP [CREATE key groupname id-or-$] [SETID key groupname id-or-$] [DESTROY key groupname] [DELCONSUMER key groupname consumername]
  • keyИмя очереди; создайте её, если она не существует
  • groupnameИмя группы.
  • $Указывает на потребление с конца, принимаются только новые сообщения. Все текущие сообщения Stream будут проигнорированы.

Начать потребление с начала:

XGROUP CREATE mystream consumer-group-name 0-0

Начать потребление с конца:

XGROUP CREATE mystream consumer-group-name $

XREADGROUP GROUP

Используйте XREADGROUP GROUP для чтения сообщений из группы потребителей. Формат синтаксиса:

XREADGROUP GROUP group consumer [COUNT count] [BLOCK milliseconds] [NOACK] STREAMS key [key ...] ID [ID ...]
  • groupИмя группы потребителей
  • consumerИмя потребителя.
  • countКоличество для чтения.
  • millisecondsМиллисекунды блокировки.
  • keyИмя очереди.
  • IDID сообщения.
XREADGROUP GROUP consumer-group-name consumer-name COUNT 1 STREAMS mystream >
Другие расширения