如何更快遍历Redis数据库中35万条键值对？

兄弟，你当前代码慢的根本问题出在35万次单独的网络请求带来的往返时间累加——每个get都是一次独立的网络调用，哪怕Redis处理每个请求只需要几微秒，35万次的“代码发请求→Redis处理→Redis返回结果”的循环延迟加起来，就会变成分钟级的耗时。这也是为什么redis-benchmark的批量测试结果和你的实际代码差这么多，因为benchmark用的是批量请求，把网络开销降到了最低。

至于线程优化，它确实能在一定程度上缩短耗时（比如并行发请求，把串行的等待时间变成并行的），但这绝对不是最优解。毕竟Redis本身是单线程处理命令的，多个线程发过来的请求最终还是会被Redis排队串行执行，而且线程池的创建、切换以及连接池的管理都会带来额外开销，实际提升可能远不如你预期，还会把代码搞复杂。

真正的最优解是减少网络请求次数，结合Redis的批量命令和渐进式扫描，具体分两步：

1. 用scan替代keys，避免阻塞Redis #

keys *命令会一次性遍历Redis所有键，在大数据库上会阻塞Redis进程，导致其他请求超时。scan是渐进式迭代，每次返回一小部分键，不会长时间阻塞服务，更适合生产环境。

2. 用mget或Pipeline批量获取值 #

• mget可以一次传入多个键，Redis一次性返回所有对应的值，把N次请求变成1次，直接把网络开销降到最低。
• 如果需要更灵活的批量操作（比如除了get还有其他命令），可以用Pipeline打包多个命令，Redis一次性执行并返回结果。

优化后的代码（scan + mget） #

import redis
import timeit

def fetch_all_data():
    start_time = timeit.default_timer()
    r = redis.Redis(host='127.0.0.1', port=6379, db=9)
    data = {}
    cursor = 0
    batch_size = 1000  # 每次扫描的键数量，可根据实际网络情况调整

    while True:
        cursor, keys = r.scan(cursor=cursor, count=batch_size)
        if not keys:
            break
        # 批量获取所有键对应的值
        values = r.mget(keys)
        # 批量处理键值对
        for key, value in zip(keys, values):
            if value:
                data[key.decode('utf-8')] = int(value.decode('utf-8'))
    
    elapsed = timeit.default_timer() - start_time
    print(f'Time to read {len(data)} records: {elapsed:.2f} seconds')
    return data

if __name__ == '__main__':
    fetch_all_data()

用Pipeline的版本（适合更复杂的批量操作） #

import redis
import timeit

def fetch_all_data_with_pipeline():
    start_time = timeit.default_timer()
    r = redis.Redis(host='127.0.0.1', port=6379, db=9)
    data = {}
    cursor = 0
    batch_size = 1000

    while True:
        cursor, keys = r.scan(cursor=cursor, count=batch_size)
        if not keys:
            break
        # 使用Pipeline打包多个get命令
        pipe = r.pipeline()
        for key in keys:
            pipe.get(key)
        # 一次性执行所有命令并获取结果
        values = pipe.execute()
        # 处理结果
        for key, value in zip(keys, values):
            if value:
                data[key.decode('utf-8')] = int(value.decode('utf-8'))
    
    elapsed = timeit.default_timer() - start_time
    print(f'Time to read {len(data)} records: {elapsed:.2f} seconds')
    return data

if __name__ == '__main__':
    fetch_all_data_with_pipeline()

这两种方案都能把请求次数从35万次降到几百次（按batch_size=1000算，就是350次左右），网络开销直接砍到原来的千分之一，耗时应该能降到几秒级别，远优于线程方案。而且scan不会阻塞Redis，更适合生产环境使用。

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