Python 操作 Redis

1. Redis 简介

Redis 是主要把数据放在内存里的键值数据库。程序通过 key 找到对应的 value，读取或修改其中的数据。 它常用于保存会话、缓存查询结果、计数和记录已处理的任务。

Redis 的读写主要在内存中完成，调用耗时还包括网络传输和命令处理。数据很大、命令处理量很大时，响应也会变慢。

Redis 支持通过 RDB 快照和 AOF 日志把数据保存到磁盘。

RDB 快照：把某时的内存数据保存成文件。重启时读入文件，恢复保存数据。

AOF 日志：记录写入、修改、删除等操作，重启时重放已保存的日志来恢复数据。日志的写盘频率由配置决定，异常退出时，尚未写入磁盘的操作可能丢失。

2. Redis 中的数据组织

2.1 key 与 value

一个 key 对应一个 value，value 有自己的数据类型。 Redis 不要求每个 key 都有相同字段，也不需要提前建表。

例如：

chat:session:s001 中的冒号是命名约定，用来区分业务、对象和编号。Redis 把整个名称当成一个 key，不会按冒号建立目录。

JSON 文本存进 String，类型仍然是 String。 其中的 history 是 JSON 数组，不能直接用 Redis 的列表命令操作。Python 需要解析这段 JSON，才能读取里面的字段。

2.2 数据库编号

单机 Redis 支持用编号区分逻辑数据库，常见的默认配置是 16 个，编号为 0～15。具体数量由服务端配置决定，Python 客户端的 db=0 表示选择 0 号库。

不同库中的同名 key 相互独立，但这些库共用同一个 Redis 实例的资源。Redis Cluster 只支持 0 号库，业务区分通常放进 key 前缀。

3. Python 连接 Redis 与连接池

3.1 安装与连接

redis-py 是 Python 的 Redis 客户端，安装包名为 redis。 它负责发送命令、接收结果，Redis 服务需要单独运行。

代码使用 Python 3.11+、redis-py 7.2.0，连接已运行的 Redis 7.4 服务。示例地址为 127.0.0.1:6379，使用时替换为实际服务地址。

1

python -m pip install "redis==7.2.0"

连接本机 Redis，写入并读取一个字符串：

1
2
3
4
5
6
7
8

import redis

client = redis.Redis(host="127.0.0.1", port=6379, db=0, decode_responses=True)
client.set("redis-demo:python", "连接成功", ex=60)
print(client.get("redis-demo:python"))
# 输出：连接成功

client.close()

decode_responses=True 把返回的字节解码成字符串。不设置时，get() 读到的文本会是 b"..."；计数、长度等结果仍返回整数。

字符串解码与 JSON 解析是两件事。 decode_responses=True 不会把 '{"turn_id": 1}' 转成字典，这一步需要 json.loads()。

3.2 连接池

连接池保存可以复用的连接，不保存查询结果。 普通命令执行时，客户端从池中取得连接；结果读完，连接归还池中。

直接创建 redis.Redis(...) 时，客户端默认已经带有连接池。显式创建池，适合集中设置参数或让多个客户端共用一个池。

pool 管理连接，r 负责执行命令：

1
2
3
4
5
6
7

import redis

pool = redis.ConnectionPool(
    host="127.0.0.1", port=6379, db=0, decode_responses=True,
    max_connections=20, socket_connect_timeout=2, socket_timeout=2,
)
r = redis.Redis(connection_pool=pool)

connection_pool=pool 指定客户端使用哪个池。

数据读写共用客户端 r，运行时把这段连接池定义与操作代码放在同一个 Python 文件中。

普通 ConnectionPool 达到上限且连接全被占用时，会抛出 MaxConnectionsError。需要等待空闲连接时，使用 BlockingConnectionPool，它的 timeout 控制等待时间。

一个池最多允许 20 条连接。20 条都被占用时，第 21 个操作使用普通池会报错；使用 BlockingConnectionPool(timeout=3)，会等待连接归还，超过 3 秒仍取不到连接则报错。

socket_timeout=2 约束一次 socket 读写等待，不代表整个业务函数一定在 2 秒内结束。排队、其他命令和重试都可能增加总耗时。

客户端通常跟随工作进程存活，不必每个请求重新创建。多个工作进程各自管理池，max_connections=20 只约束当前这个池，总连接数还要考虑进程数量。

3.3 同步调用与异步调用

redis.Redis 的网络读写会阻塞当前线程。放进 async def 中调用，这一点也不会改变。

事件循环负责调度同一线程里的多个协程。异步客户端等待 Redis 响应时，当前协程暂停，这个线程能继续处理其他协程；同步客户端等待时，线程会停在当前调用上。

异步代码使用 redis.asyncio：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

import asyncio
import redis.asyncio as aioredis


async def main():
    client = aioredis.Redis(host="127.0.0.1", port=6379, decode_responses=True)
    try:
        await client.set("redis-demo:async", "异步调用", ex=60)
        print(await client.get("redis-demo:async"))
        # 输出：异步调用
    finally:
        await client.aclose()


asyncio.run(main())

3.4 单机与 Cluster

单机客户端使用 redis.Redis；Cluster 客户端使用 redis.cluster.RedisCluster，由客户端发现节点并路由命令。

Cluster 中涉及多个 key 的命令和事务通常要求这些 key 落在同一个哈希槽。

4. 常用数据类型

4.1 String：文本、数字与 JSON

set() 写入整个值，get() 取回整个值。对已有 key 调用 set() 会覆盖原值；没有设置过期参数时，原来的过期时间会被清除。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

r.set("redis-demo:name", "小林", ex=60)
print(r.get("redis-demo:name"))
# 输出：小林

r.set("redis-demo:visits", 10)
print(r.incr("redis-demo:visits"))
# 11

print(r.incrby("redis-demo:visits", 5))
# 16

print(r.get("redis-demo:visits"))
# 16

incr() 加 1，incrby() 增加指定整数，返回 Python 整数。被操作的值必须是可解析的整数，修改在 Redis 内完成。计数仍以 String 保存，get() 读到的是字符串 "16"。

set() 的常用参数：

mset({"key1": "value1", "key2": "value2"}) 批量写入，mget(["key1", "key2"]) 按传入顺序返回值列表。

4.2 Hash：按字段保存对象

Hash 在一个 key 下面保存多个字段。 hset() 接收字段名和值，mapping 参数用于一次写入多个字段。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

key = "redis-demo:user:u001"
r.delete(key)

print(r.hset(key, mapping={"name": "小林", "city": "上海"}))
# 输出：2

print(r.hget(key, "city"))
# 上海

print(r.hset(key, "city", "杭州"))
# 0

print(r.hget(key, "city"))
# 杭州

hset() 返回新增字段数。第二次写 city 返回 0，是因为这个字段已经存在，值仍然会更新。

hgetall(key) 返回包含全部字段和值的 Python 字典，hdel(key, "city") 删除指定字段。

写入字段值时，Python 字典或列表需要序列化，例如转成 JSON 字符串。

String 中的 JSON 适合整体读写，Hash 适合经常单独读取或修改字段的对象。

4.3 List：有顺序的记录

List 的元素有位置，内容允许重复。 lpush() 从左侧插入，rpush() 从右侧插入；lpop()、rpop() 取出并删除对应一端的元素。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

key = "redis-demo:messages"
r.delete(key)

print(r.rpush(key, "你好", "想了解配置"))
# 输出：2

print(r.lrange(key, 0, -1))
# ['你好', '想了解配置']

print(r.lpop(key))
# 你好

print(r.llen(key))
# 1

rpush() 返回插入后的长度，lrange() 返回 Python 列表。起止下标都包含在结果内，lrange(key, 0, -1) 读取全部元素，lrange(key, 0, 49) 读取前 50 个。

ltrim(key, -50, -1) 只保留最后 50 个元素。使用 rpush() 追加消息时，保留的就是最近写入的 50 条。

4.4 Set：成员判断与去重

Set 中同一个成员只保存一份。 它不记录插入顺序，适合判断某个编号是否已出现。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

key = "redis-demo:used"
r.delete(key)

print(r.sadd(key, "opening_a", "opening_b", "opening_a"))
# 输出：2

print(r.sismember(key, "opening_a"))
# 1

print(r.scard(key))
# 2

print(r.srem(key, "opening_b"))
# 1

sadd() 返回新增成员数，重复的 opening_a 不再添加。sismember() 返回 1 表示存在、0 表示不存在；scard() 返回成员总数，srem() 返回移除的成员数。

smembers(key) 返回 Python 集合，遍历顺序不固定。

4.5 Sorted Set：带分数的集合

Sorted Set 的成员不重复，每个成员对应一个分数。 zadd() 接收“成员 → 分数”的字典，同一个成员再次写入时会更新分数。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

key = "redis-demo:ranking"
r.delete(key)

print(r.zadd(key, {"user_a": 80, "user_b": 95, "user_c": 88}))
# 输出：3

print(r.zrange(key, 0, 1, desc=True, withscores=True))
# [('user_b', 95.0), ('user_c', 88.0)]

print(r.zscore(key, "user_b"))
# 95.0

print(r.zrevrank(key, "user_b"))
# 0

zrange() 默认按分数从小到大取数据，desc=True 改为从大到小。0, 1 取前两名，withscores=True 让结果同时带上分数。

zrank() 查询升序名次，zrevrank() 查询倒序名次，两者都从 0 开始。分数相同时按成员的字节字典序排列，倒序查询会反转这个顺序。zrem(key, "user_b") 删除成员。

5. 实操

5.1 会话的读写

会话保存轮次 turn_id 和历史消息 history，整体转成 JSON 字符串写入 Redis。

load_session() 读取会话；save_session() 接收 Python 字典，负责序列化和写入。

SESSION_TTL 是代码中定义的会话有效期，单位为秒。这里设为 72 * 60 * 60，即 72 小时。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

import json

SESSION_TTL = 72 * 60 * 60


def load_session(session_id):
    key = f"redis-demo:session:{session_id}"
    raw = r.get(key)
    if raw is None:
        return {"turn_id": 0, "history": []}
    session = json.loads(raw)
    # 读到已有会话，把剩余有效期重新设为 72 小时
    r.expire(key, SESSION_TTL)
    return session


def save_session(session_id, session):
    key = f"redis-demo:session:{session_id}"
    # 写入会话，同时把有效期设为 72 小时
    return r.set(key, json.dumps(session, ensure_ascii=False), ex=SESSION_TTL)


r.delete("redis-demo:session:s001")
session = load_session("s001")
session["turn_id"] += 1
session["history"].append({"role": "user", "content": "你好"})
save_session("s001", session)

print(load_session("s001"))
# 输出：{'turn_id': 1, 'history': [{'role': 'user', 'content': '你好'}]}

ensure_ascii=False 让 JSON 中的中文保持原样；不设置这个参数，中文会写成转义形式，仍能正常还原。json.loads() 把 JSON 字符串还原为字典和列表。

key 不存在时，load_session() 返回初始会话；调用 save_session() 才写入 Redis。

读取失败与没有历史是两回事。 连接失败或 JSON 解析失败时，函数会抛出异常，由调用方处理。

5.2 会话有效期

TTL 表示 key 的剩余有效期。

新建 key 时，如果没有设置过期时间，默认不会过期，r.ttl(key) 返回 -1。

访问时重新设置有效期，称为滑动过期。 持续被访问的会话可能长期存在。

假设会话写入时 TTL 为 72 小时，期间没有续期。48 小时后读取，剩余有效期为 24 小时。读取函数调用 expire()，把它重新设为 72 小时。

get() 不会续期，expire(key, seconds) 为已有 key 设置有效期。set(key, value, ex=seconds) 把写入和设置有效期放在一次 Redis 操作中完成。

hset()、lpush() 会保留已有 key 的 TTL，不会把剩余时间重新计满。

过期的 key 再用 get() 读取会返回 None，会话函数因此返回初始状态。主动删除、内存淘汰也会让 key 提前消失。

JSON 会话的历史越长，每次整体读写的数据就越大。TTL 控制保留时长，历史条数和字段大小决定会话体积。

5.3 并发覆盖

会话函数包含读取、Python 中修改、写回三个动作。一次 set() 对应的写入不会被其他命令拆开执行，但多个动作之间可能穿插其他请求。

两个请求同时修改同一会话，可能互相覆盖：

消息 A 被覆盖，是因为两个请求都基于旧值修改。把保存函数改成异步，这种覆盖仍可能发生。

同一会话的请求按顺序处理时，需要等一个请求完成读取、修改和写回，另一个请求才能开始。多个工作进程共同处理请求时，单个进程内的排队无法约束其他进程。

只追加消息时，List 的 rpush() 能在 Redis 内完成一次追加。根据读取结果修改数据时，WATCH 事务或 Lua 用来处理读写之间的并发冲突。上述会话函数采用普通读写，没有加入这些处理。

5.4 去重标记

给事件 event001 设置一个保留 180 秒的处理标记：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

key = "redis-demo:dedup:event001"
# 清除上次演示的标记，不放进事件处理函数
r.delete(key)

accepted = r.set(key, "1", nx=True, ex=180)
print(accepted)
# 输出：True

accepted_again = r.set(key, "1", nx=True, ex=180)
print(accepted_again)
# None

nx=True 只在 key 不存在时写入。 第一次返回 True，重复写入返回 None。业务代码仅在 accepted is True 时继续处理事件；180 秒后 key 过期，新请求能再次写入。

因 key 已存在而跳过写入时，不会刷新 TTL。若业务处理失败，标记仍然存在，其他请求在过期前会被跳过。标记只表示已被领取，不表示处理完成。

5.5 批量执行命令

需要写入多条独立数据时，pipeline 把命令集中发送，减少多次请求之间的网络往返。

1
2
3
4
5
6
7
8

with r.pipeline(transaction=False) as pipe:
    pipe.set("redis-demo:batch:a", "A", ex=60)
    pipe.set("redis-demo:batch:b", "B", ex=60)
    pipe.get("redis-demo:batch:a")
    results = pipe.execute()

print(results)
# 输出：[True, True, 'A']

pipe.set() 记录待执行命令，execute() 才发送。结果列表与命令顺序对应，这里的两个 True 对应写入，'A' 对应读取。

transaction=False 只做批量发送，不把这组命令作为事务执行。 单机 redis-py 的 pipeline() 默认启用事务，事务执行期间不会穿插其他客户端的命令。

Redis 事务不提供执行错误回滚：其中一条命令执行时报错，其他命令仍可能执行，已经完成的修改不会自动撤销。

5.6 结束连接

所有使用方结束操作时，关闭客户端并断开池内连接：

1
2

r.close()
pool.disconnect()

r.close() 不会断开通过 connection_pool=pool 传入的共享池，断开连接由 pool.disconnect() 完成。多个客户端共用一个池时，要等它们都结束操作。