Python如何从Redis中读取哈希键及对应TIMESTAMP字段值

注意事项 #

r.keys() 命令会遍历全库所有键，执行期间会阻塞Redis主线程，键量较大时会严重影响实例性能，仅建议测试环境、小数据集场景使用。生产环境请使用下文的SCAN迭代方案。

基础实现（直接遍历键列表） #

适合小数据集测试场景：

# r为已完成初始化的Redis连接实例
intkeylist = r.keys()
result = []

for key in intkeylist:
    # 过滤非哈希类型的键，避免调用hget报错
    if r.type(key) == b"hash":
        ts_value = r.hget(key, "TIMESTAMP")
        # 字节转字符串（redis-py默认返回字节类型，若已配置decode_responses=True可跳过decode步骤）
        key_str = key.decode("utf-8")
        ts_value_str = ts_value.decode("utf-8") if ts_value else "无TIMESTAMP字段"
        # 存入结果变量
        result.append((key_str, "TIMESTAMP", ts_value_str))
        # 按要求格式输出到屏幕
        print(f"{key_str} TIMESTAMP {ts_value_str}")

优化实现（Pipeline批量请求） #

使用Redis Pipeline批量提交请求，减少网络往返开销，键量中等时性能提升明显：

intkeylist = r.keys()
pipe = r.pipeline(transaction=False)

for key in intkeylist:
    pipe.hget(key, "TIMESTAMP")
# 批量执行所有查询请求
ts_values = pipe.execute()

result = []
for key, ts_value in zip(intkeylist, ts_values):
    if r.type(key) == b"hash":
        key_str = key.decode("utf-8")
        ts_value_str = ts_value.decode("utf-8") if ts_value else "无TIMESTAMP字段"
        result.append((key_str, "TIMESTAMP", ts_value_str))
        print(f"{key_str} TIMESTAMP {ts_value_str}")

生产环境实现（SCAN迭代遍历） #

使用SCAN命令增量迭代键，不会阻塞Redis实例，适合大库生产场景：

result = []
cursor = 0
# 每次迭代扫描的键数量，可根据实际业务调整
scan_count = 1000

while True:
    cursor, keys = r.scan(cursor, count=scan_count)
    for key in keys:
        if r.type(key) == b"hash":
            ts_value = r.hget(key, "TIMESTAMP")
            key_str = key.decode("utf-8")
            ts_value_str = ts_value.decode("utf-8") if ts_value else "无TIMESTAMP字段"
            result.append((key_str, "TIMESTAMP", ts_value_str))
            print(f"{key_str} TIMESTAMP {ts_value_str}")
    # 游标归0代表全库遍历完成
    if cursor == 0:
        break

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