Python脚本如何预防缓存雪崩问题

wen python案例 2026-06-26 243

本文目录导读：

Python脚本如何预防缓存雪崩问题

设置随机过期时间
缓存预热（预加载）
互斥锁（加锁单线程重建缓存）
使用Redis分布式锁
多级缓存策略
限流与降级
综合应用示例

缓存雪崩是指大量缓存数据在同一时间过期失效，导致所有请求直接打到数据库，造成数据库压力骤增甚至崩溃,Python脚本预防缓存雪崩的常用策略如下：

设置随机过期时间

避免所有缓存使用相同的过期时间,在基础过期时间上增加随机偏移：

import random
import time
def get_random_expiry(base_ttl=3600):
    """基础过期时间上增加随机偏移（0-600秒）"""
    return base_ttl + random.randint(0, 600)
# 使用示例
cache.set(key, value, ex=get_random_expiry())

缓存预热（预加载）

在系统低峰期提前加载热点数据到缓存：

def warm_up_cache():
    """系统启动或低峰期预加载热点数据"""
    hot_keys = get_hot_keys_from_db()  # 从数据库获取热点key
    for key in hot_keys:
        if not cache.exists(key):
            value = db.query(key)  # 从数据库获取数据
            cache.set(key, value, ex=get_random_expiry())

互斥锁（加锁单线程重建缓存）

当缓存失效时，只允许一个线程去重建缓存,其他线程等待：

import threading
class CacheManager:
    def __init__(self):
        self.locks = {}
        self.lock = threading.Lock()
    def get_with_lock(self, key, rebuild_func, ttl=3600):
        # 尝试从缓存获取数据
        data = cache.get(key)
        if data is not None:
            return data
        # 获取或创建key对应的锁
        with self.lock:
            if key not in self.locks:
                self.locks[key] = threading.Lock()
            key_lock = self.locks[key]
        # 加锁重建缓存
        with key_lock:
            # 双重检查，防止其他线程已重建
            data = cache.get(key)
            if data is not None:
                return data
            # 从数据库重建缓存
            data = rebuild_func()
            cache.set(key, data, ex=ttl)
            return data

使用Redis分布式锁

对于分布式系统，使用Redis的SETNX实现分布式锁：

import redis
import time
r = redis.Redis()
def get_with_distributed_lock(key, rebuild_func, ttl=3600, lock_timeout=10):
    """使用分布式锁防止缓存雪崩"""
    data = r.get(key)
    if data:
        return data
    lock_key = f"lock:{key}"
    lock_value = str(time.time())
    # 尝试获取锁，设置过期时间防止死锁
    if r.setnx(lock_key, lock_value):
        r.expire(lock_key, lock_timeout)
        try:
            # 双重检查
            data = r.get(key)
            if data:
                return data
            # 重建缓存
            data = rebuild_func()
            r.setex(key, ttl, data)
            return data
        finally:
            # 释放锁（只释放自己的锁）
            if r.get(lock_key) == lock_value:
                r.delete(lock_key)
    else:
        # 获取锁失败，等待重试
        time.sleep(0.1)
        return get_with_distributed_lock(key, rebuild_func, ttl, lock_timeout)

多级缓存策略

使用本地缓存（如LRU）配合Redis缓存：

from functools import lru_cache
import redis
r = redis.Redis()
class MultiLevelCache:
    def __init__(self, local_ttl=300, redis_ttl=3600):
        self.local_ttl = local_ttl
        self.redis_ttl = redis_ttl
    # 本地缓存（一级缓存，过期时间短）
    @lru_cache(maxsize=1000)
    def get_local(self, key):
        return None  # 实际实现需要存储时间和值
    def get(self, key, rebuild_func):
        # 1. 尝试本地缓存
        data = self.get_local(key)
        if data:
            return data
        # 2. 尝试Redis缓存
        data = r.get(key)
        if data:
            self.set_local(key, data)  # 回写本地缓存
            return data
        # 3. 从数据库获取
        data = rebuild_func()
        # 4. 写入各级缓存
        self.set_local(key, data, self.local_ttl)
        r.setex(key, self.redis_ttl, data)
        return data

限流与降级

控制同时重建缓存的并发数,必要时降级：

from contextlib import contextmanager
import threading
class CircuitBreaker:
    def __init__(self, max_concurrent=5):
        self.semaphore = threading.Semaphore(max_concurrent)
        self.is_open = False
    @contextmanager
    def acquire(self):
        if self.is_open:
            raise Exception("Circuit breaker is open")
        acquired = self.semaphore.acquire(blocking=False)
        if not acquired:
            raise Exception("Too many concurrent requests")
        try:
            yield
        finally:
            self.semaphore.release()
circuit_breaker = CircuitBreaker()
def safe_get_data(key, rebuild_func):
    try:
        with circuit_breaker.acquire():
            return rebuild_func()
    except Exception:
        # 降级处理
        return get_stale_data(key)  # 返回过期数据或默认值

综合应用示例

import random
import time
from functools import wraps
def anti_cache_avalanche(ttl=3600, lock_timeout=10):
    """装饰器实现缓存雪崩防护"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            key = f"{func.__name__}:{args}:{kwargs}"
            # 启用熔断机制
            if hasattr(wrapper, 'breaker') and wrapper.breaker.is_open:
                return get_fallback_value(key)  # 返回降级数据
            # 尝试获取缓存（带随机过期）
            data = cache.get(key)
            if data:
                return data
            # 使用互斥锁重建缓存
            lock_key = f"lock:{key}"
            if cache.setnx(lock_key, 1, ex=lock_timeout):
                try:
                    # 双重检查
                    data = cache.get(key)
                    if data:
                        return data
                    # 执行原始函数获取数据
                    data = func(*args, **kwargs)
                    # 设置缓存（随机过期时间）
                    actual_ttl = ttl + random.randint(0, 600)
                    cache.set(key, data, ex=actual_ttl)
                    return data
                finally:
                    cache.delete(lock_key)
            else:
                # 等待并重试
                time.sleep(0.1)
                return wrapper(*args, **kwargs)
        return wrapper
    return decorator
# 使用示例
@anti_cache_avalanche(ttl=3600)
def get_user_info(user_id):
    # 从数据库获取用户信息
    return db.query(f"SELECT * FROM users WHERE id={user_id}")