34. Python 客户端 redis-py 实战

Python 是运维、数据分析与后端服务里最常用的语言之一，操作 Redis 的首选项就是官方维护的 redis-py。它的接口风格和 redis-cli 几乎一一对应：SET 对应 r.set、GET 对应 r.get，学习成本极低。本章码上学带你从安装、连接、基本读写一路走到 Pipeline 批处理，并讲清楚 redis-py 背后连接池的默认行为，为第 35 章的性能调优埋下伏笔。

127.0.0.1:6379> SET greeting "hello redis"
OK
127.0.0.1:6379> GET greeting
"hello redis"
127.0.0.1:6379> HSET user:1 name tom age 20
(integer) 2
127.0.0.1:6379> HGETALL user:1
1) "name"
2) "tom"
3) "age"
4) "20"

本节目标

• 安装 redis-py 并建立到 Redis 8.x 的连接，理解 decode_responses 的作用。
• 用 redis-py 完成字符串、列表、哈希的读写与遍历。
• 掌握 pipeline() 批处理，明白它如何把多次网络往返压缩成一次。
• 理解 redis-py 默认就带连接池，知道为什么多线程下通常要给每个线程单独建客户端实例。
• 能写出一段同时具备「正确性」与「性能意识」的 Python 访问代码。

一、安装与连接

用 pip 安装（版本以 PyPI 最新稳定版为准）：

pip install redis

建立连接时，最常用的参数是 host、port、db、password，以及一个非常关键的 decode_responses：

import redis

# decode_responses=True 让返回值是 str 而不是 bytes
r = redis.Redis(
    host='localhost',
    port=6379,
    db=0,
    password=None,          # 若 Redis 设了密码，填在这里
    decode_responses=True
)

print("连接成功:", r.ping())   # 对应 redis-cli 的 PING，返回 True

decode_responses=True 强烈建议打开：否则 r.get('greeting') 返回的是 b'hello redis' 这种字节串，每次都要手动 .decode()，既麻烦又容易在拼接字符串时踩坑。只有当你确实需要存二进制数据（图片、序列化对象）时，才考虑保持默认的 False。

二、基本读写

字符串

r.set('greeting', 'hello redis')
print(r.get('greeting'))          # hello redis

# 带过期时间（单位：秒）
r.set('code', '8423', ex=60)
print(r.ttl('code'))              # 剩余秒数

列表

LPUSH 从左压入，LRANGE 取区间，-1 表示到末尾：

r.lpush('tutorial:list', 'Redis', 'Mongodb', 'Mysql')
print(r.lrange('tutorial:list', 0, -1))   # ['Mysql', 'Mongodb', 'Redis']

哈希

HSET 可一次写入多个字段（redis-py 用 mapping= 参数），HGETALL 返回字典：

r.hset('user:1', mapping={'name': 'tom', 'age': '20'})
print(r.hgetall('user:1'))        # {'name': 'tom', 'age': '20'}
print(r.hget('user:1', 'name'))   # tom

遍历键

和命令行一样，KEYS 会全库扫描、阻塞服务，学习阶段可用，生产请用 scan_iter：

# 学习/调试用，生产慎用
for key in r.keys('*'):
    print(key)

# 生产推荐：游标分批扫描，不阻塞
for key in r.scan_iter(match='user:*', count=100):
    print(key)

批量读写

除了 Pipeline，redis-py 还提供了直接的批量命令：mset 一次写多个键、mget 一次读多个键，它们在底层各自只产生一次网络往返，适合「一次性读一组相关键」的场景：

r.mset({'user:1:name': 'tom', 'user:1:age': '20'})
print(r.mget('user:1:name', 'user:1:age'))   # ['tom', '20']
print(r.exists('user:1:name', 'missing'))     # 存在键的数量：1

mxet/mget 与 pipeline 的区别在于：前者是协议内建的批量命令、语义固定；后者可以任意混搭不同命令并获得各自返回值，灵活性更高。日常「读多个独立键」用 mget 最直观，复杂批处理再上 pipeline。

三、Pipeline 批处理

当你要连续执行很多命令时，如果每发一条等一条回复，网络往返（RTT）会严重拖慢速度。redis-py 的 pipeline() 把多条命令攒在一起，最后一次 execute() 提交：

pipe = r.pipeline()
for i in range(1000):
    pipe.set(f'item:{i}', i)
# 一次性发送并取回全部结果
results = pipe.execute()
print("共执行命令:", len(results))   # 1000

你也可以混搭不同命令，并且用 transaction=False（默认）做纯批处理，或在 pipeline(transaction=True) 下让整批以事务方式执行：

pipe = r.pipeline()
pipe.set('a', 1)
pipe.incr('a')
pipe.get('a')
print(pipe.execute())   # [True, 2, '2']

这里 execute() 返回的是每条命令回复组成的列表，顺序与 pipe 调用顺序一致。SET 返回 True，INCR 返回整数 2，GET 返回字符串 '2'。

四、关于连接池

一个容易忽略的点：redis-py 的 Redis 对象内部已经自带连接池。每次执行命令时，它从池里借一个连接，用完归还。这意味着你创建一次 r = redis.Redis(...) 后反复调用方法是没问题的，底层会自动复用连接。

但在多线程场景下要注意：虽然 redis-py 客户端本身是线程安全的（靠内部连接池保证），但如果你把同一个 pipeline() 对象在多个线程间共享，就可能出问题——Pipeline 是有状态的。正确做法是每个线程各自 r.pipeline()，而 r 这个客户端实例可以共用。

import threading

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

def worker(n):
    # 每个线程用自己的 pipeline，客户端 r 可共享
    pipe = r.pipeline()
    pipe.incr('counter')
    pipe.execute()

threads = [threading.Thread(target=worker, args=(i,)) for i in range(10)]
for t in threads:
    t.start()
for t in threads:
    t.join()

print("最终计数:", r.get('counter'))   # 10

如果你想显式控制连接池的大小（比如限制最多 16 个连接），可以自己创建并传入：

pool = redis.ConnectionPool(
    host='localhost',
    port=6379,
    db=0,
    decode_responses=True,
    max_connections=16
)
r = redis.Redis(connection_pool=pool)

max_connections 默认是 2 ** 31，基本等于不限制，所以一般不用改；但在连接数容易失控的场景（如 serverless 函数、频繁新建客户端）显式限制更安全。

五、异常处理与连接形式

真实网络环境下，连接可能由于服务端重启、网络抖动或 timeout 配置而被断开。用 redis-py 时应捕获 redis.exceptions 下的异常做兜底：

import redis

r = redis.Redis(host='localhost', port=6379, decode_responses=True)
try:
    r.set('k', 'v')
    print(r.get('k'))
except redis.exceptions.TimeoutError:
    print("连接或读取超时，请检查网络与服务端负载")
except redis.exceptions.ConnectionError:
    print("无法连接到 Redis，请检查地址、端口与密码")

redis-py 具备一定的韧性：连接断开后，下一次命令会自动尝试重建连接（底层连接池会重新分配），所以一般无需手写重连循环。但要注意，如果服务端 timeout 把空闲连接踢掉，正在进行的那条命令可能失败——对此加上上面的异常捕获即可。

除了 host/port，还有两种常见的连接形式：

# 用 URI 一步建连（含密码、库号）
r = redis.from_url("redis://:password@localhost:6379/0", decode_responses=True)

# 走 Unix 域套接字（同机部署、省去 TCP 开销）
r = redis.Redis(unix_socket_path="/tmp/redis.sock", decode_responses=True)

小结

• redis-py 是 Python 操作 Redis 8.x 的官方推荐客户端，命令名与 redis-cli 基本一一对应。
• 连接时打开 decode_responses=True 可避免到处处理 bytes，是工程上的省心之选。
• 基本读写覆盖字符串、列表、哈希；生产环境遍历键请用 scan_iter 而非 keys。
• pipeline() 把多条命令合并成一次网络往返，是提升吞吐的关键手段；需要原子性时加 transaction=True。
• redis-py 默认内置连接池，多线程应共享客户端实例、各用各的 Pipeline；需要限制连接数时显式传入 ConnectionPool。

常见误区

• 忘了 decode_responses，拿到 b'...' 后直接字符串拼接报错。统一在连接时设 decode_responses=True 最省事。
• 在多线程里共享同一个 pipeline() 对象。Pipeline 有状态，应每个线程独立创建。
• 用 keys('*') 在生产环境全量遍历。大数据量下会阻塞 Redis，务必改用 scan_iter。