使用Python操作Redis的完整指南

pip install redis 

import redis
# 连接本地Redis
r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
# 测试连接
r.set('foo', 'bar')
print(r.get('foo'))  # 输出: bar

pool = redis.ConnectionPool(host='localhost', port=6379, db=0, max_connections=10)
r = redis.Redis(connection_pool=pool)

def get_user(user_id):
    # 直接查数据库
    return db.query(f"SELECT * FROM users WHERE id={user_id}")

def get_user(user_id):
    # 先查缓存
    cache_key = f"user:{user_id}"
    user = r.get(cache_key)
    
    if user:
        return json.loads(user)  # 缓存命中，直接返回
    
    # 缓存未命中，查数据库
    user = db.query(f"SELECT * FROM users WHERE id={user_id}")
    
    # 写入缓存，设置过期时间
    r.setex(cache_key, 3600, json.dumps(user))
    return user

from functools import wraps
import json

def redis_cache(ttl=300):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # 生成缓存key：函数名 + 参数
            key = f"{func.__name__}:{str(args)}:{str(kwargs)}"
            
            # 尝试从缓存获取
            cached = r.get(key)
            if cached:
                return json.loads(cached)
            
            # 执行原函数
            result = func(*args, **kwargs)
            
            # 写入缓存
            r.setex(key, ttl, json.dumps(result))
            return result
        return wrapper
    return decorator

# 使用
@redis_cache(ttl=600)
def get_user(user_id):
    return db.query(f"SELECT * FROM users WHERE id={user_id}")

def get_user(user_id):
    user = r.get(f"user:{user_id}")
    if user is not None:  # 注意：None表示缓存未命中，空字符串表示缓存了空值
        return user if user != "NULL" else None
    
    user = db.query(...)
    # 无论查没查到，都写缓存
    r.setex(f"user:{user_id}", 600, user or "NULL")
    return user

def get_hot_data(key):
    data = r.get(key)
    if data:
        return data
    
    # 加锁，只允许一个线程去查数据库
    with r.lock(f"lock:{key}", timeout=10):
        # 双重检查：拿到锁后可能已经被其他线程更新了
        data = r.get(key)
        if data:
            return data
            
        data = expensive_query()
        r.setex(key, 3600, data)
        return data

import random

# 基础过期时间3600秒，加上0-300秒的随机偏移
expire = 3600 + random.randint(0, 300)
r.setex(key, expire, value)

from redis._cache import _LocalCache, EvictionPolicy

# 初始化本地缓存：最多存10000条，30秒过期，LRU淘汰策略
local_cache = _LocalCache(max_size=10000, ttl=30, eviction_policy=EvictionPolicy.LRU)

def get_user_with_multilevel_cache(user_id):
    # 构造命令元组作为缓存key
    command = ("GET", f"user:{user_id}")
    
    # 查本地缓存
    cached = local_cache.get(command)
    if cached:
        return cached
    
    # 查Redis
    user = r.get(f"user:{user_id}")
    if user:
        # 写入本地缓存
        local_cache.set(command, user, keys_in_command=[f"user:{user_id}"])
        return user
    
    # 查数据库
    user = db.query(...)
    r.setex(f"user:{user_id}", 3600, user)
    return user