我要评论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': 'john@example.com'
})
print(r.hgetall('user:1000')) # 输出: {b'name': b'john', b'age': b'30', b'email': b'john@example.com'}
# 获取单个字段
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可以显著提升应用性能，但也要注意数据一致性和内存管理等问题。
到此这篇关于python操作redis从基础到高级应用的文章就介绍到这了,更多相关python操作redis内容请搜索代码网以前的文章或继续浏览下面的相关文章希望大家以后多多支持代码网！
发表评论