Redis Streams извлекает несколько значений с помощью XREAD

Question

У меня есть поток redis "mystream", и я пытаюсь использовать:

XREAD COUNT 5 БЛОК 50000 ПОТОКОВ mystream $

прочитать пять новых записей, добавленных в mystream с блокировкой на 50 секунд. Как только я выполняю команду для добавления данных в мой поток, используя:

XADD mystream a 5 b 6

заблокированная команда завершается с предоставлением единственной записи в качестве вывода.

Есть ли какой-либо способ команды XREAD, с помощью которой я могу получать несколько записей, добавленных с этого момента до 50-х годов, или мне нужно вызывать XREAD несколько раз для достижения того же?

Answer 1

То, что вы просите, не предназначено для использования XREAD BLOCK. Я бы предложил команду XREVRANGE:

xrevrange mystream + - COUNT 5

min: "+" - указывает наивысший идентификатор потока, а max: "-" - указывает на самый низкий идентификатор потока. Вместо "-" вы также можете передать TIMESTAMP, который ограничит вашXREVRANGE результаты к желаемому времени TIMESTAMP секунды назад (включительно).

xrevrange mystream + 1596617691455 COUNT 5

Предложение выше не дает того же blockingповедение. Если вы все еще хотите реагировать на входящие данные, вы можете заглянуть в Redis Pub / Sub. Подпишитесь на тему и напишите свою собственную логику для обработки входящих данных. Вот пример веб-чата.

Вы также можете использовать XREAD BLOCK чтобы определить минимальное значение вашего XREVRANGE команда:

XREAD COUNT 5 BLOCK 50000 STREAMS mystream $
XREVRANGE mystream + [returned_xread_key] COUNT 5

Извините, но я не могу привести примеры на языке C, вот на Python:

import redis
import time
redis_client = redis.StrictRedis("0.0.0.0", decode_responses=True)
response = redis_client.xread({'mystream': "$"}, count=1, block=50000)
if len(response) > 0:
    min_value = response[0][1][0][0] # [['mystream', [('1596618943439-0', {'mykey': 'myval'})]]]
    time.sleep(50) # wait for the end of the window
    entries = redis_client.xrevrange('mystream', max="+", min=min_value, count=5)