Python脚本如何操作Redis缓存

wen 实用脚本 2026-07-12 1

Python脚本操作Redis缓存：从入门到精通

📚 目录导读

Redis与Python：黄金搭档的基础认知
环境搭建：安装与连接
核心操作：增删改查实战
高级玩法：管道、事务与发布订阅
性能优化：连接池与序列化
常见问题与解决方案（Q&A）
最佳实践与避坑指南

Redis与Python：黄金搭档的基础认知

在现代Web开发与数据密集型应用中，Redis作为高性能的键值存储系统，常被用于缓存、会话管理、消息队列等场景，Python凭借其简洁的语法和丰富的生态,成为操作Redis的首选语言之一。

Python脚本如何操作Redis缓存

Redis核心特性：

内存级读写，毫秒级响应
支持多种数据结构（String、Hash、List、Set、Sorted Set）
提供过期时间、持久化、主从复制等企业级功能

Python操作Redis的方式： 官方推荐使用redis-py库（最新版本为redis 5.x），它提供了同步和异步两种接口,完美适配不同场景。

环境搭建：安装与连接

1 安装Redis服务器

# macOS
brew install redis
# Ubuntu/Debian
sudo apt-get install redis-server
# 启动服务
redis-server

2 安装Python客户端

pip install redis
# 如需异步支持
pip install redis[hiredis]

3 建立连接

import redis
# 基础连接
r = redis.Redis(host='localhost', port=6379, db=0, password=None)
# 支持连接池（推荐用于生产环境）
pool = redis.ConnectionPool(host='localhost', port=6379, decode_responses=True)
r = redis.Redis(connection_pool=pool)
# 测试连接
print(r.ping())  # 返回True则连接成功

注意：decode_responses=True会自动将字节数据解码为字符串,避免手动decode。

核心操作：增删改查实战

1 字符串操作（最常用）

# 设置
r.set('username', 'tech_writer')  
# 设置带过期时间（5秒）
r.setex('session_token', 5, 'abc123')  
# 批量设置
r.mset({'key1': 'value1', 'key2': 'value2'})
# 获取
print(r.get('username'))  # b'tech_writer'
# 批量获取
print(r.mget(['key1', 'key2']))  # [b'value1', b'value2']
# 删除
r.delete('key1')
# 检查是否存在
print(r.exists('username'))  # 1

2 Hash操作（适合对象缓存）

# 设置单个字段
r.hset('user:1001', 'name', 'Alice')
# 批量设置
r.hset('user:1001', mapping={'age': 30, 'email': 'alice@example.com'})
# 获取所有字段
print(r.hgetall('user:1001'))  
# 获取单个字段
print(r.hget('user:1001', 'name'))  # b'Alice'

3 列表与集合

# 列表：右侧推入
r.rpush('tasks', 'task1', 'task2')
# 左侧弹出
task = r.lpop('tasks')
# 集合：自动去重
r.sadd('tags', 'python', 'redis', 'python')  # 只存一个python
members = r.smembers('tags')  # 返回所有元素

4 过期时间管理

# 设置键的过期时间（秒）
r.expire('username', 3600)
# 查看剩余时间
ttl = r.ttl('username')  # -1表示永不过期

高级玩法：管道、事务与发布订阅

1 管道（Pipeline）：批量操作性能优化

当需要发送多条命令时，管道可以将多条命令打包发送,减少网络往返时间。

pipe = r.pipeline()
pipe.set('key1', 'val1')
pipe.set('key2', 'val2')
pipe.expire('key1', 60)
pipe.execute()  # 一次性执行所有命令

2 事务（Transaction）：保证原子性

pipe = r.pipeline(transaction=True)
try:
    pipe.watch('balance')  # 监控键
    current_balance = int(pipe.get('balance'))
    pipe.multi()
    pipe.set('balance', current_balance - 100)
    pipe.execute()
except redis.WatchError:
    print("键被修改，事务回滚")

3 发布订阅：实时消息系统

# 发布者
pub = redis.Redis()
pub.publish('channel:news', 'New article published')
# 订阅者
sub = redis.Redis()
p = pubsub()
p.subscribe('channel:news')
for message in p.listen():
    if message['type'] == 'message':
        print(f"收到消息: {message['data'].decode()}")

性能优化：连接池与序列化

1 连接池管理

# 设置最大连接数
pool = redis.ConnectionPool(max_connections=50, 
                            socket_timeout=5,
                            socket_connect_timeout=3)
r = redis.Redis(connection_pool=pool)
# 使用后无需手动关闭，连接池自动回收

2 序列化策略

对于复杂对象（字典、列表）,建议使用JSON或Pickle序列化后存储。

import json
# 存储对象
user_data = {"name": "Bob", "scores": [90, 85, 88]}
r.set('user:bob', json.dumps(user_data))
# 读取对象
loaded = json.loads(r.get('user:bob'))

3 缓存过期策略

# 设置缓存失效时间（随机过期防止雪崩）
def set_cache_with_jitter(key, value, base_ttl=300):
    import random
    jitter = random.uniform(0, 60)
    r.setex(key, base_ttl + jitter, value)

常见问题与解决方案（Q&A）

❓ 问题1：连接Redis时出现“Connection refused”

原因： Redis服务未启动或端口被占用。
解决： 检查服务状态 redis-cli ping，确保redis-server正在运行。

❓ 问题2：获取的数据总是bytes类型

原因： 默认返回字节串。
解决： 创建连接时添加decode_responses=True参数，或手动调用.decode()。

❓ 问题3：如何防止缓存穿透？

方案： 使用布隆过滤器（Bloom Filter）。

from pybloom_live import BloomFilter
bf = BloomFilter(capacity=100000, error_rate=0.001)
# 查询前检查布隆过滤器
if key in bf:
    value = r.get(key)
else:
    # 从数据库查询并更新缓存

❓ 问题4：大规模数据插入很慢怎么办？

方案： 使用管道批量提交，每次提交1000条左右。

pipe = r.pipeline()
for i in range(10000):
    pipe.set(f'key:{i}', f'value:{i}')
    if i % 1000 == 0:
        pipe.execute()
pipe.execute()  # 处理剩余数据

❓ 问题5：生产环境如何保证高可用？

方案： 使用Redis Sentinel或Redis Cluster。

# Sentinel模式
from redis.sentinel import Sentinel
sentinel = Sentinel([('sentinel1', 26379), ('sentinel2', 26379)])
master = sentinel.master_for('mymaster', socket_timeout=0.5)
slave = sentinel.slave_for('mymaster', socket_timeout=0.5)

最佳实践与避坑指南

✅ 必须要做的配置

设置合理的过期时间：避免缓存永久占用内存
使用连接池：避免每次请求创建新连接
监控慢查询：通过Redis的SLOWLOG命令分析性能瓶颈

❌ 常见错误避免

不要在生产环境使用KEYS命令：会阻塞Redis，改用SCAN游标命令
避免将Redis当作数据库：Redis是缓存，数据安全性依赖持久化配置
注意数据序列化开销：超大对象建议拆分存储

📈 性能提升技巧

使用hiredis解析器：比纯Python解析器快10倍
批量操作优先使用管道和MSET
对于热点数据，考虑本地缓存+Redis二级缓存架构

延伸阅读：

Redis官方文档：redis.io/documentation
redis-py GitHub仓库：github.com/redis/redis-py
布隆过滤器实现参考：pybloom-live库