本文目录导读：
我来为您详细介绍Python连接Redis缓存的几种常见方式：
基础安装
pip install redis
基本连接方式
1 基础连接
import redis
# 方式1：直接连接（默认连接本地6379端口）
r = redis.Redis(host='localhost', port=6379, db=0)
# 方式2：使用URL连接
r = redis.from_url('redis://localhost:6379/0')
# 测试连接
print(r.ping()) # 返回True表示连接成功
2 带密码的连接
# 有密码认证
r = redis.Redis(
host='localhost',
port=6379,
password='your_password',
db=0,
decode_responses=True # 自动解码返回为字符串
)
连接池方式（推荐）
import redis
# 创建连接池
pool = redis.ConnectionPool(
host='localhost',
port=6379,
password='your_password',
db=0,
max_connections=10, # 最大连接数
decode_responses=True
)
# 从连接池获取连接
r = redis.Redis(connection_pool=pool)
完整案例：用户缓存系统
import redis
import json
import time
class RedisCache:
def __init__(self):
# 初始化连接池
self.pool = redis.ConnectionPool(
host='localhost',
port=6379,
db=0,
decode_responses=True
)
self.client = redis.Redis(connection_pool=self.pool)
def set_user(self, user_id, user_data, expire=3600):
"""设置用户缓存"""
key = f"user:{user_id}"
# 将字典转为JSON字符串存储
value = json.dumps(user_data)
# 设置过期时间（秒）
return self.client.setex(key, expire, value)
def get_user(self, user_id):
"""获取用户缓存"""
key = f"user:{user_id}"
value = self.client.get(key)
if value:
return json.loads(value)
return None
def delete_user(self, user_id):
"""删除用户缓存"""
key = f"user:{user_id}"
return self.client.delete(key)
def is_logged_in(self, user_id):
"""检查用户是否在线"""
key = f"session:{user_id}"
return self.client.exists(key)
# 使用示例
def main():
cache = RedisCache()
# 设置用户缓存
user_info = {
"id": 1001,
"name": "张三",
"email": "zhangsan@example.com",
"login_time": time.time()
}
cache.set_user(1001, user_info)
# 获取用户缓存
cached_user = cache.get_user(1001)
print(f"缓存用户信息: {cached_user}")
# 删除缓存
cache.delete_user(1001)
if __name__ == "__main__":
main()
Redis Sentinel（哨兵模式）
from redis.sentinel import Sentinel
# 连接哨兵
sentinel = Sentinel([
('sentinel-host1', 26379),
('sentinel-host2', 26379),
], socket_timeout=0.1)
# 获取主节点连接
master = sentinel.master_for('mymaster', socket_timeout=0.1)
# 获取从节点连接（用于读操作）
slave = sentinel.slave_for('mymaster', socket_timeout=0.1)
# 使用主节点写入
master.set('key', 'value')
# 使用从节点读取
value = slave.get('key')
Redis Cluster（集群模式）
from rediscluster import RedisCluster
# 安装：pip install redis-py-cluster
startup_nodes = [
{"host": "127.0.0.1", "port": "7000"},
{"host": "127.0.0.1", "port": "7001"},
{"host": "127.0.0.1", "port": "7002"}
]
# 创建集群连接
rc = RedisCluster(startup_nodes=startup_nodes, decode_responses=True)
# 使用示例
rc.set("foo", "bar")
print(rc.get("foo"))
常用操作方法
# 字符串操作
r.set('key', 'value')
r.get('key')
r.setex('key', 60, 'value') # 设置过期时间60秒
# 列表操作
r.lpush('list', 'a', 'b', 'c')
r.rpush('list', 'd', 'e')
r.lrange('list', 0, -1) # 获取所有元素
# 哈希操作
r.hset('hash', 'field', 'value')
r.hget('hash', 'field')
r.hgetall('hash')
# 集合操作
r.sadd('set', 'a', 'b', 'c')
r.smembers('set') # 获取所有元素
# 有序集合
r.zadd('zset', {'a': 1, 'b': 2, 'c': 3})
r.zrange('zset', 0, -1, withscores=True)
异常处理
import redis
from redis.exceptions import ConnectionError, TimeoutError
class RedisClient:
def __init__(self):
self.client = None
self.connect()
def connect(self):
try:
self.client = redis.Redis(
host='localhost',
port=6379,
socket_connect_timeout=2, # 连接超时
socket_timeout=2, # 操作超时
retry_on_timeout=True, # 超时重试
decode_responses=True
)
self.client.ping()
print("Redis连接成功")
except (ConnectionError, TimeoutError) as e:
print(f"Redis连接失败: {e}")
def get_data(self, key):
try:
return self.client.get(key)
except Exception as e:
print(f"获取数据失败: {e}")
return None
注意事项
- 连接池管理：生产环境务必使用连接池
- 异常处理：做好连接超时、异常重试机制
- 数据序列化：复杂对象需要序列化（JSON、pickle等）
- 内存管理：合理设置过期时间，避免内存溢出
- 安全认证：生产环境要设置密码，避免暴露在公网
这些是Python连接Redis的主流方式,您可以根据实际场景选择合适的方案。
