Redis - xadd 和 xread 的 Python 示例

Question

您能否给出一个在 Python 中使用 Redis 的非常简单的示例 xread 和 xadd （显示 xread 的返回值的类型和格式以及 xadd 的输入）？我已经阅读了很多文档，但没有一个是用 Python 编写的。

Redis 文档 给出了一个例子：

 > XADD mystream * sensor-id 1234 temperature 19.8
1518951480106-0

但如果我在 python 中尝试：

 sample = {b"hello":b"12"}
id = r.xadd("mystream", sample)

我收到此错误：

 redis.exceptions.ResponseError: WRONGTYPE Operation against a key holding the wrong kind of value

原文由 AgnosticCucumber 发布，翻译遵循 CC BY-SA 4.0 许可协议

Answer 1

确保在运行之前进行刷新以确保不存在具有相同名称的列表/流。 :

 redis-cli flushall

 if __name__ == '__main__':
    r = redis.Redis(host='localhost', port=6379, db=0)
    encoder = JSONEncoder()
    sample = {"hello": encoder.encode([1234,125, 1235, 1235])} # converts list to string
    stream_name = 'mystream'

    for i in range(10):
        r.xadd(stream_name, sample)

    # "$" doesn't seem to work in python
    read_samples = r.xread({stream_name:b"0-0"})

原文由 AgnosticCucumber 发布，翻译遵循 CC BY-SA 4.0 许可协议

Answer 2

基于 redis-py 文档：

Redis初始化：

 redis = redis.Redis(host='localhost')

添加键值对（键值应该是一个字典）：

 redis.xadd(stream_name, {key: value})

阻止阅读：

 redis.xread({stream_name: '$'}, None, 0)

1. stream_name 和 ID 应该是一个字典。
2. $ 表示最新消息。

此外，我没有为流 mystream 传递普通 ID，而是传递了特殊 ID $。这个特殊的 ID 意味着 XREAD 应该使用已经存储在流 mystream 中的最大 ID 作为最后一个 ID，这样我们将只接收新消息，从我们开始监听的时间开始。 从这里

1. 如果您想接收最新消息，而不仅仅是任意数量的消息，则 COUNT 应该为 NONE。
2. BLOCK 选项为 0 表示 Block 超时为 0 毫秒（这意味着永远不会超时）

原文由 Mo MahdiM 发布，翻译遵循 CC BY-SA 4.0 许可协议