Redis重启后缓存全失怎么兜底？预热和降级策略怎么设计？-php教程-PHP中文网

Redis重启后缓存全失怎么兜底？

通过设置合理的过期时间并结合应用层的多级缓存或数据库直读机制，可以有效防止redis重启导致的缓存穿透和雪崩。具体做法是：在写入缓存时设置较短的过期时间（如5-10分钟），避免长时间无更新；同时，在应用层实现“缓存未命中 -> 查db -> 回填缓存”的逻辑，并加入分布式锁或本地缓存（如caffeine）来防止并发击穿。此外，可配置redis的aof持久化策略，确保数据在重启后能部分恢复。

import redis
import time

class CacheManager:
    def __init__(self):
        self.redis_client = redis.Redis(host='localhost', port=6379, db=0)
        self.local_cache = {}  # 简易本地缓存作为二级兜底

    def get_data(self, key):
        # 1. 先查本地缓存
        if key in self.local_cache:
            return self.local_cache[key]

        # 2. 再查Redis
        data = self.redis_client.get(key)
        if data:
            decoded_data = data.decode('utf-8')
            self.local_cache[key] = decoded_data  # 同步到本地缓存
            return decoded_data

        # 3. Redis未命中，查数据库（模拟）
        db_data = self._query_database(key)
        if db_data:
            # 4. 回填Redis，设置短过期时间
            self.redis_client.setex(key, 300, db_data)  # 5分钟过期
            self.local_cache[key] = db_data
            return db_data

        return None

    def _query_database(self, key):
        # 模拟数据库查询
        return f"data_for_{key}" if key else None

# 使用示例
cm = CacheManager()
print(cm.get_data("user_123"))

预热和降级策略怎么设计？

预热策略应在系统启动或流量低谷期主动加载热点数据到Redis，而降级策略则在Redis不可用时自动切换至数据库直连或返回默认值。具体实现包括：在应用启动时异步加载预定义的热点Key集合；在Redis连接失败时，通过熔断器（如Hystrix或Resilience4j）快速失败，并 fallback 到数据库查询或静态默认响应。同时，监控Redis的健康状态，动态调整降级阈值。

import redis
import threading
import time
from functools import lru_cache

class PreloadAndFallback:
    def __init__(self):
        self.redis_client = redis.Redis(host='localhost', port=6379, db=0)
        self.hot_keys = ["hot_item_1", "hot_item_2"]  # 预定义热点Key
        self.fallback_enabled = False

    def preload_hot_data(self):
        """预热热点数据"""
        for key in self.hot_keys:
            try:
                data = self._fetch_from_db(key)
                self.redis_client.setex(key, 600, data)  # 预热，10分钟过期
                print(f"Preloaded: {key}")
            except Exception as e:
                print(f"Failed to preload {key}: {e}")

    def get_with_fallback(self, key):
        """带降级的获取逻辑"""
        if self.fallback_enabled:
            return self._fallback_to_db(key)

        try:
            data = self.redis_client.get(key)
            if data:
                return data.decode('utf-8')
        except redis.ConnectionError:
            self.fallback_enabled = True  # 触发降级
            return self._fallback_to_db(key)

        return None

    def _fetch_from_db(self, key):
        return f"db_value_{key}"

    def _fallback_to_db(self, key):
        return self._fetch_from_db(key)

# 启动预热
preload_thread = threading.Thread(target=PreloadAndFallback().preload_hot_data)
preload_thread.start()