如何使用Python解析Redis Streams消息并提取字典内容

先明确xread返回值的层级逻辑 #

结合你实际拿到的返回样例[['streamA', [('1611593593691-0', {'identifier': 'zxkbdnarrz', 'query': 'testQuery'})]]]，每一层的含义固定：

• 最外层列表：对应你本次调用xread监听的所有流的返回集合，因为你仅监听streamA，列表长度固定为1
• 二级元素为固定格式二元组：结构是[流名称, 当前流下的消息列表]
• 三级的消息列表内，每个元素又是固定二元组：结构是(消息全局ID, 消息体键值对)
• 元组的第二个元素就是目标消息内容，因为你的Redis连接开启了decode_responses=True，这个元素本身就是可直接操作的标准Python字典，不需要额外做格式转换。

可直接落地的提取代码 #

不要用硬索引直接取值（比如fromStreamA[0][1][0][1]），一旦xread阻塞超时返回空值，硬索引会直接触发IndexError崩溃，推荐用遍历的方式做容错提取：

# Redis连接及xread调用代码同你原有实现
r1 = redis.Redis(host=redis_host, port=redis_port,
             password=redis_password, db=1, decode_responses=True)
fromStreamA = r1.xread({'streamA': "$"}, count=1, block=0)

# 增加非空判断，适配block超时返回空的场景
if fromStreamA:
    # 遍历所有返回的流（单流场景下循环仅执行1次）
    for stream_name, messages in fromStreamA:
        # 遍历当前流下的所有消息（count=1场景下循环仅执行1次）
        for message_id, message_body in messages:
            # 直接从字典取值即可完成后续业务逻辑
            current_identifier = message_body["identifier"]
            current_query = message_body["query"]
            # 后续可直接对接JSON序列化、业务逻辑判断等操作，无额外格式转换成本

常见注意点 #

• 不需要额外调用json.loads解析消息：你发送的是Redis Hash结构消息，xread读取时会直接将Hash的键值对映射为Python字典，不存在JSON字符串嵌在返回值里的情况
• 你当前用的'$'作为流读取起点，含义是仅读取调用xread之后新产生的消息，不会回溯历史消息，和阻塞监听的场景匹配
• 如果后续需要扩展监听多个Stream，上述提取逻辑不需要修改，只需要在xread的入参里增加对应流名和读取起点即可

内容的提问来源于stack exchange，提问作者arthem