Redis Stream¶

Тип данных Redis Stream¶

Описание¶

В Redis реализован и доступен для бесплатного использования тип данных Redis Stream которая ведёт себя как append-only log. Из RedisStream можно читать содержимое, по аналогии с командой tail -n, tail -f.

Список команд RedisStream¶

№	Команда	Краткое описание
1	XACK	Подтвердить обработку одного или нескольких сообщений в группе потребителей.
2	XADD	Добавить новую запись в поток.
3	XAUTOCLAIM	Автоматически забрать «зависшие» сообщения из PEL и назначить их указанному потребителю, возвращая также курсор для следующего вызова.
4	XCLAIM	Переназначить «зависшие» сообщения другому потребителю внутри группы.
5	XDEL	Удалить одну или несколько записей из потока.
6	XLEN	Получить количество записей в потоке.
7	XPENDING	Показать сводку и/или список «зависших» сообщений в группе (владельцы, счетчики доставок, диапазоны).
8	XRANGE	Получить диапазон записей по ID в порядке возрастания.
9	XREVRANGE	Получить диапазон записей по ID в порядке убывания.
10	XREAD	Прочитать записи из одного или нескольких потоков (возможна блокировка до появления новых записей).
11	XREADGROUP	Прочитать записи от имени группы потребителей.
12	XTRIM	Обрезать поток до заданной длины или минимального ID (опции MAXLEN/MINID, LIMIT).
13	XINFO CONSUMERS	Информация о потребителях в указанной группе потока.
14	XINFO GROUPS	Информация о группах потребителей для потока.
15	XINFO STREAM	Информация о самом потоке (включая FULL/COUNT).
16	XINFO HELP	Справка по подкомандам XINFO.
17	XGROUP CREATE	Создать группу потребителей для потока (с опцией MKSTREAM при необходимости).
18	XGROUP CREATECONSUMER	Создать потребителя в существующей группе.
19	XGROUP DELCONSUMER	Удалить потребителя из группы (очистив его PEL).
20	XGROUP DESTROY	Удалить группу потребителей.
21	XGROUP SETID	Установить/изменить last-delivered-id для группы.
22	XGROUP HELP	Справка по подкомандам XGROUP.
23	XSETID	Установить последний сгенерированный ID потока без добавления записи.

Создадим окружение¶

Redis¶

Для запуска Redis локально, я использую redis-stack.

Если контенера локально нет, то необходимо спуллить и запустить

docker pull redis/redis-stack
docker run -d --name redis-stack -p 6379:6379 -p 8001:8001 redis/redis-stack:latest

Если контейнер уже создавался и погашен, то просто запустить

docker container start redis-stack

Python¶

Необходимые зависимости:

redis==6.4.0

Создадим и заполним стрим¶

In [14]:

Copied!





from dataclasses import dataclass, asdict
import redis

r = redis.Redis()
key = 'llm-stream:42144aaa-6e56-4e8e-be97-ca17bd6c99e1'

@dataclass
class Message:
    number: int
    name: str = 'Тестовое сообщение'

message = Message(number=0)
from dataclasses import dataclass, asdict
import redis

r = redis.Redis()
key = 'llm-stream:42144aaa-6e56-4e8e-be97-ca17bd6c99e1'

@dataclass
class Message:
    number: int
    name: str = 'Тестовое сообщение'

message = Message(number=0)

In [24]:

Copied!

for number in range(1, 11):
    message.number = number
    r.xadd(key, asdict(message), id=f'0-{number}')
for number in range(1, 11):
    message.number = number
    r.xadd(key, asdict(message), id=f'0-{number}')

In [25]:

Copied!

print(r.xlen(key))
print(r.xlen(key))

In [ ]: