Python Redis案例如何连接Redis缓存

wen python案例 2026-06-30 253

本文目录导读：

基础安装
基本连接方式
连接池方式（推荐）
完整案例：用户缓存系统
Redis Sentinel（哨兵模式）
Redis Cluster（集群模式）
常用操作方法
异常处理
注意事项

我来为您详细介绍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