Python操作Redis核心方法：连接池、Pipeline与五大类型详解

脚本专家

本文聚焦 Python 通过 redis-py 库操作 Redis 的核心技术，从环境准备到连接管理、五大数据类型 API、Pipeline 批量加速以及异常重试机制，提供可直接落地的代码实践。适合需要将 Redis 集成到 Python 后端、缓存系统或任务队列的开发人员。

一、环境准备与安装
redis-py 是 Redis 官方推荐的 Python 客户端，纯 Python 实现，API 与 redis-cli 命令几乎一一对应。安装命令：

2. pip install redis

复制代码

安装后可使用 pip show redis 检查版本。本文以 redis-py 5.x 为例，建议使用 4.0 以上版本以获得更好的类型提示和异步支持。

二、创建连接：从单机到生产规范
2.1 基础连接
最简单的连接方式如下，decode_responses=True 开关必须开启，否则返回值会以 bytes 形式出现，需要手动解码。

2. import redis
3. r = redis.Redis(host='localhost', port=6379, decode_responses=True)
4. print(r.ping())  # True

复制代码

2.2 带密码和数据库编号的连接
通过 db 参数指定数据库编号（默认 16 个数据库，0-15）。

2. r = redis.Redis(
3.     host='192.168.1.100',
4.     port=6379,
5.     password='your_password',
6.     db=1,
7.     decode_responses=True
8. )

复制代码

2.3 使用 URL 连接（现代推荐）
从配置文件中读取连接信息时，URL 格式更方便。注意密码前加冒号。

2. import os
3. r = redis.from_url('redis://:mypassword@localhost:6379/0', decode_responses=True)
4. print(r.ping())
5. # 永远不要硬编码密码，使用环境变量
6. r = redis.from_url(os.getenv('REDIS_URL'))

复制代码

三、五大基础数据类型 API 速查
redis-py 的方法名与 redis-cli 命令一致，参数顺序相同。以下逐一演示。
3.1 String 操作

2. r = redis.Redis(decode_responses=True)
3. r.set('name', 'Alice')
4. print(r.get('name'))  # Alice
5. r.mset({'age': 30, 'city': 'Beijing'})
6. print(r.mget(['age', 'city']))  # ['30', 'Beijing']
8. # 设置过期时间（ex = 秒）、仅当键不存在时设置（nx）
9. r.set('token', 'abc', ex=10)
10. r.set('lock', '1', nx=True, ex=30)
12. # 数字操作
13. r.set('counter', 0)
14. r.incr('counter')
15. r.incrby('counter', 5)
16. print(r.get('counter'))  # 6
18. # 字符串操作
19. r.set('greet', 'Hello')
20. r.append('greet', ' Redis')
21. print(r.get('greet'))  # Hello Redis
22. print(r.strlen('greet'))  # 11
23. print(r.getrange('greet', 0, 4))  # Hello

复制代码

3.2 Hash 操作

2. r.hset('user:1001', mapping={'name': 'Bob', 'age': '25', 'city': 'Shanghai'})
3. print(r.hgetall('user:1001'))  # {'name': 'Bob', 'age': '25', 'city': 'Shanghai'}
4. print(r.hget('user:1001', 'name'))  # Bob
5. print(r.hmget('user:1001', ['name', 'age']))  # ['Bob', '25']
6. r.hincrby('user:1001', 'age', 1)
7. print(r.hget('user:1001', 'age'))  # 26
8. print(r.hexists('user:1001', 'city'))  # True
9. r.hdel('user:1001', 'city')
10. print(r.hlen('user:1001'))  # 2

复制代码

3.3 List 操作

2. r.lpush('queue', 'task1', 'task2')  # queue: ['task2', 'task1']
3. r.rpush('queue', 'task3')          # queue: ['task2', 'task1', 'task3']
4. print(r.lrange('queue', 0, -1))    # ['task2', 'task1', 'task3']
5. print(r.lpop('queue'))  # task2
6. print(r.rpop('queue'))  # task3
7. # 阻塞弹出：r.brpop('queue', timeout=5)  # 返回 (key, value) 或 None
8. print(r.llen('queue'))  # 1
9. print(r.lindex('queue', 0))  # task1
10. r.lpush('log', *['a','b','c','d','e'])
11. r.ltrim('log', 0, 2)
12. print(r.lrange('log', 0, -1))  # ['e', 'd', 'c']

复制代码

3.4 Set 操作

2. r.sadd('tags:python', 'redis', 'django', 'flask')
3. print(r.smembers('tags:python'))  # {'flask', 'django', 'redis'}（无序）
4. print(r.sismember('tags:python', 'django'))  # True
5. print(r.scard('tags:python'))  # 3
6. r.srem('tags:python', 'flask')
8. # 集合运算
9. r.sadd('tags:web', 'django', 'fastapi')
10. print(r.sinter('tags:python', 'tags:web'))  # {'django'}
11. print(r.sunion('tags:python', 'tags:web'))  # {'redis', 'django', 'fastapi'}
12. print(r.sdiff('tags:python', 'tags:web'))  # {'redis'}
13. print(r.spop('tags:python'))  # 随机弹出一个元素

复制代码

3.5 Sorted Set 操作

2. r.zadd('leaderboard', {'Alice': 100, 'Bob': 85, 'Charlie': 92})
3. print(r.zrange('leaderboard', 0, -1, withscores=True))
4. # [('Bob', 85.0), ('Charlie', 92.0), ('Alice', 100.0)]
5. print(r.zrevrange('leaderboard', 0, 1, withscores=True))
6. # [('Alice', 100.0), ('Charlie', 92.0)]
7. r.zincrby('leaderboard', 10, 'Bob')
8. print(r.zscore('leaderboard', 'Bob'))  # 95.0
9. print(r.zrank('leaderboard', 'Bob'))   # 0（从低到高排名）
10. print(r.zrevrank('leaderboard', 'Bob')) # 2（从高到低排名）
11. print(r.zrangebyscore('leaderboard', 90, 100))
12. # ['Charlie', 'Alice']

复制代码

四、连接池管理：告别频繁握手
每次新建 Redis 连接都需要 TCP 三次握手和认证，高并发场景下会浪费资源并可能超过 maxclients 限制。连接池预先创建一组连接，请求时借、用完还，连接复用。

2. pool = redis.ConnectionPool(
3.     host='localhost',
4.     port=6379,
5.     decode_responses=True,
6.     max_connections=20  # 池中最大连接数
7. )
8. r1 = redis.Redis(connection_pool=pool)
9. r2 = redis.Redis(connection_pool=pool)
10. print(r1.ping())
11. print(r2.ping())
12. pool.disconnect()  # 应用退出时释放所有连接

复制代码

生产环境建议封装为工厂函数，从环境变量读取配置，并设置超时和健康检查：

2. import os
3. def get_redis_client():
4.     pool = redis.ConnectionPool(
5.         host=os.getenv('REDIS_HOST', 'localhost'),
6.         port=int(os.getenv('REDIS_PORT', 6379)),
7.         password=os.getenv('REDIS_PASSWORD', None),
8.         db=int(os.getenv('REDIS_DB', 0)),
9.         decode_responses=True,
10.         max_connections=int(os.getenv('REDIS_MAX_CONNS', 50)),
11.         socket_timeout=5,
12.         socket_connect_timeout=5,
13.         health_check_interval=30,
14.     )
15.     return redis.Redis(connection_pool=pool)

复制代码

注意：在多线程环境下应全局共享一个连接池和一个 Redis 实例（redis-py 的 Redis 对象是线程安全的）。多进程（如 gunicorn）则每个 worker 进程创建自己的连接池。

五、Pipeline：批量命令的加速器
每次 Redis 命令都要经历一次网络往返（RTT）。Pipeline 将多个命令打包一次性发送，大幅减少网络开销。性能对比：

2. import time
3. r = redis.Redis(decode_responses=True)
5. # 无 Pipeline：1000 次 SET
6. start = time.perf_counter()
7. for i in range(1000):
8.     r.set(f'key:{i}', f'value:{i}')
9. print(f'无 Pipeline 耗时: {time.perf_counter() - start:.3f}s')
11. # 使用 Pipeline
12. r.flushdb()
13. start = time.perf_counter()
14. pipe = r.pipeline()
15. for i in range(1000):
16.     pipe.set(f'key:{i}', f'value:{i}')
17. pipe.execute()
18. print(f'使用 Pipeline 耗时: {time.perf_counter() - start:.3f}s')

复制代码

本地测试中 Pipeline 耗时往往仅为循环写法的几十分之一。Pipeline 还支持链式调用并获取每个命令的返回结果：

2. pipe = r.pipeline()
3. pipe.set('name', 'Alice').incr('counter').get('name')
4. results = pipe.execute()
5. print(results)  # [True, 1, 'Alice']

复制代码

如果需要原子性事务，可在 Pipeline 中启用 transaction=True：

2. pipe = r.pipeline(transaction=True)
3. pipe.set('balance', 100)
4. pipe.incrby('balance', 50)
5. pipe.execute()  # 在 MULTI/EXEC 中执行

复制代码

最佳实践：一次 Pipeline 中的命令数量建议控制在 1000 以内，避免客户端内存溢出和长时间阻塞 Redis 主线程。Pipeline 会独占一个连接，执行完毕后归还连接池。

六、异常处理与重试机制
网络抖动时，需要捕获 ConnectionError 和 TimeoutError 并实现重试。手工实现退避重试：

2. from redis.exceptions import ConnectionError, TimeoutError
3. import time
4. def safe_incr(r, key, retries=3):
5.     for attempt in range(retries):
6.         try:
7.             return r.incr(key)
8.         except (ConnectionError, TimeoutError) as e:
9.             print(f'第 {attempt+1} 次重试: {e}')
10.             time.sleep(0.1 * (attempt + 1))
11.     raise Exception(f'操作失败，已重试 {retries} 次')

复制代码

redis-py 4.x 以上版本也内置了 retry_on_timeout 和 retry_on_error 参数，可直接在连接时配置：

2. r = redis.Redis(
3.     host='localhost',
4.     retry_on_timeout=True,
5.     retry_on_error=[ConnectionError, TimeoutError],
6.     health_check_interval=30,
7. )

复制代码

七、实战练习建议
1. 连接池挑战：创建 max_connections=5 的连接池，启动 10 个线程同时执行 GET 操作，观察线程是否阻塞，验证连接复用行为。
2. Pipeline 性能测试：准备 5000 个键值对，分别用 for 循环 SET 和 Pipeline 写入，再分别用 GET 循环和 MGET 读取，记录耗时。
3. 缓存读写函数：编写 get_cached_data(key)，先从 Redis 读取，不存在时模拟从数据库查询并写入 Redis（带 TTL），第二次调用应命中缓存。

八、总结
本文完整覆盖了 Python 通过 redis-py 操作 Redis 的核心技术：安装、连接（包括 URL 方式）、五大数据类型 API、连接池管理、Pipeline 批量加速以及异常重试。掌握这些内容后，你已能写出工程级的 Redis 交互逻辑。后续可深入事务、Lua 脚本、序列化以及异步 redis.asyncio 等进阶主题。

热心网友5

很实用的总结！redis-py 的 API 和原生命令几乎一一对应，确实降低了学习成本。日常开发中强烈建议开启 `decode_responses=True`，不然 bytes 处理起来很麻烦。 有一点想请教：帖子标题里提到了 Pipeline 和异常重试，但正文只展示了五大数据类型的操作，能否补充一段 Pipeline 的批量写入示例，以及生产环境里常见的连接断开重试处理方式？这样会更完整，对新手也更有帮助。