Python操作Redis：从基础到高级应用
|
🌺The Begin🌺点点关注，收藏不迷路🌺
|
1. 环境准备与安装
1.1 安装Redis-Py库
pip install redis
对于异步支持（Python 3.7+）：
pip install redis[asyncio]
1.2 连接Redis服务器
基础连接示例：
import redis
# 创建连接池（推荐）
pool = redis.ConnectionPool(host='localhost', port=6379, db=0, password='yourpassword')
r = redis.Redis(connection_pool=pool)
# 简单连接
r = redis.Redis(host='localhost', port=6379, db=0)
2. 基础数据类型操作
2.1 字符串(String)操作
# 设置和获取
r.set('name', 'Alice')
print(r.get('name')) # 输出: b'Alice'
# 批量操作
r.mset({'key1': 'value1', 'key2': 'value2'})
print(r.mget('key1', 'key2')) # 输出: [b'value1', b'value2']
# 自增操作
r.set('counter', 1)
r.incr('counter')
print(r.get('counter')) # 输出: b'2'
2.2 列表(List)操作
# 列表操作
r.lpush('tasks', 'task1', 'task2')
r.rpush('tasks', 'task3')
print(r.lrange('tasks', 0, -1)) # 输出: [b'task2', b'task1', b'task3']
# 弹出元素
task = r.lpop('tasks')
print(task) # 输出: b'task2'
2.3 哈希(Hash)操作
# 哈希表操作
r.hset('user:1000', mapping={
'name': 'John',
'age': '30',
'email': '[email protected]'
})
print(r.hgetall('user:1000')) # 输出: {b'name': b'John', b'age': b'30', b'email': b'[email protected]'}
# 获取单个字段
print(r.hget('user:1000', 'name')) # 输出: b'John'
2.4 集合(Set)操作
# 集合操作
r.sadd('tags', 'python', 'redis', 'database')
print(r.smembers('tags')) # 输出: {b'python', b'redis', b'database'}
# 集合运算
r.sadd('tags2', 'python', 'java')
print(r.sinter('tags', 'tags2')) # 输出: {b'python'}
2.5 有序集合(ZSet)操作
# 有序集合
r.zadd('rankings', {'player1': 100, 'player2': 85, 'player3': 95})
print(r.zrevrange('rankings', 0, 1)) # 输出: [b'player1', b'player3']
3. 高级功能应用
3.1 事务处理
# 事务示例
pipe = r.pipeline()
pipe.set('tx_key1', 'value1')
pipe.set('tx_key2', 'value2')
pipe.execute() # 提交事务
3.2 发布订阅模式
# 发布端
r.publish('news', 'Breaking news!')
# 订阅端
pubsub = r.pubsub()
pubsub.subscribe('news')
for message in pubsub.listen():
if message['type'] == 'message':
print(f"收到消息: {message['data']}")
break
3.3 Lua脚本执行
# Lua脚本示例
script = """
local current = redis.call('GET', KEYS[1])
local new = current + ARGV[1]
redis.call('SET', KEYS[1], new)
return new
"""
counter = r.eval(script, 1, 'mycounter', 5)
print(counter) # 输出增量后的值
4. 性能优化技巧
4.1 连接池配置
pool = redis.ConnectionPool(
max_connections=50,
host='localhost',
port=6379,
decode_responses=True # 自动解码为字符串
)
r = redis.Redis(connection_pool=pool)
4.2 管道(Pipeline)批量操作
# 管道批量操作
with r.pipeline() as pipe:
for i in range(1000):
pipe.set(f'key:{i}', f'value:{i}')
pipe.execute() # 一次性提交所有命令
4.3 连接流程图
5. 实战案例：缓存装饰器
def redis_cache(ttl=60):
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
# 生成唯一缓存键
cache_key = f"{func.__name__}:{args}:{frozenset(kwargs.items())}"
# 尝试从缓存获取
cached = r.get(cache_key)
if cached is not None:
return json.loads(cached)
# 缓存未命中，执行函数
result = func(*args, **kwargs)
# 设置缓存
r.setex(cache_key, ttl, json.dumps(result))
return result
return wrapper
return decorator
# 使用示例
@redis_cache(ttl=300)
def get_user_profile(user_id):
# 模拟数据库查询
time.sleep(2)
return {"id": user_id, "name": f"User{user_id}", "score": 85}
6. 常见问题解决方案
6.1 连接超时处理
from redis.exceptions import TimeoutError
try:
r.ping()
except TimeoutError:
print("Redis连接超时")
# 重连逻辑
r = redis.Redis(host='localhost', socket_timeout=5)
6.2 大Key问题检测
# 检测大Key
big_keys = r.execute_command('MEMORY USAGE', 'some_large_key')
if big_keys > 1024 * 1024: # 大于1MB
print(f"警告: 大Key detected - {big_keys} bytes")
6.3 集群模式支持
from redis.cluster import RedisCluster
rc = RedisCluster(
startup_nodes=[
{"host": "127.0.0.1", "port": "7000"},
{"host": "127.0.0.1", "port": "7001"}
],
decode_responses=True
)
rc.set("cluster_key", "value")
7. 最佳实践总结
- 连接管理：始终使用连接池，避免频繁创建/关闭连接
- 数据序列化：使用JSON或Msgpack等格式存储复杂对象
- 错误处理：实现健壮的重试机制和降级策略
- 性能监控：定期检查慢查询和大Key
- 合理过期：为缓存数据设置适当的TTL
8. 扩展资源
- 官方文档：redis-py GitHub
- 异步客户端：aioredis
- ORM集成：Django-Redis
通过本指南，您应该已经掌握了Python操作Redis的核心技术。合理使用Redis可以显著提升应用性能，但也要注意数据一致性和内存管理等问题。
|
🌺The End🌺点点关注，收藏不迷路🌺
|
转载自 CSDN-专业IT技术社区
原文链接：https://blog.csdn.net/qq_41840843/article/details/148644569
