缓存穿透击穿雪崩

docker run -d --name redis8 -p 6379:6379 redis:8
docker exec -it redis8 redis-cli

用户请求: GET /item/999999 (不存在)
  → Redis: MISS    (缓存不存「不存在」)
  → MySQL: MISS    (没有这行)
  → 返回空，什么也不缓存
下一个请求继续打穿……

import json, redis

r = redis.Redis(decode_responses=True)
NULL_SENTINEL = "__NULL__"

def get_item(item_id: int):
    key = f"item:{item_id}"
    data = r.get(key)
    if data is not None:
        return None if data == NULL_SENTINEL else json.loads(data)

    db_data = db.query_item(item_id)
    if db_data:
        r.setex(key, 3600, json.dumps(db_data))   # 命中缓存 1 小时
    else:
        r.setex(key, 60, NULL_SENTINEL)           # 空值只缓存 60 秒
    return db_data

# Redis 8 内置布隆过滤器，开箱即用
BF.RESERVE items:bf 0.001 10000000     # 目标误判率 0.1%，预计 1000 万元素
BF.ADD items:bf "item:1001"
BF.EXISTS items:bf "item:1001"         # 1：可能存在
BF.EXISTS items:bf "item:999999"       # 0：一定不存在，请求到此为止
BF.INFO items:bf                       # 查看实际容量与误判率

位数组 + k 个哈希函数

添加 x:  h1(x) % m、h2(x) % m、h3(x) % m 三处位置全部置 1
查询 y:  三个位置只要有一个 0 → 一定不存在（无漏报）
         三个位置全 1 → 可能存在（有误判，误判率见下）

误判率: P ≈ (1 - e^(-kn/m))^k
  m: 位数组长度   k: 哈希函数个数   n: 已插入元素数

def get_item_v2(item_id: int):
    if not r.bf().exists("items:bf", str(item_id)):
        return None                      # 一定不存在，数据库零压力

    data = r.get(f"item:{item_id}")      # 走正常缓存流程
    if data is not None:
        return None if data == NULL_SENTINEL else json.loads(data)
    db_data = db.query_item(item_id)
    if db_data:
        r.setex(f"item:{item_id}", 3600, json.dumps(db_data))
    return db_data

热点 key: hot:item:1 (TTL=3600s)
T=3600s: key 过期
  1000 个并发同时 MISS
  → 1000 个请求同时回源数据库

import time

def get_hot(key):
    data = r.get(key)
    if data:
        return json.loads(data)

    lock_key = f"lock:{key}"
    got_lock = r.set(lock_key, "1", nx=True, ex=5)   # 5 秒锁超时兜底
    if got_lock:
        try:
            data = db.query_hot(key)
            r.setex(key, 3600, json.dumps(data))
            return data
        finally:
            r.delete(lock_key)
    # 没抢到锁：稍等片刻再读缓存（回源者马上会写回）
    time.sleep(0.05)
    return get_hot_with_limit(key, depth=3)          # 递归要有层数上限

import threading, time, json

def get_hot_logical(key):
    raw = r.get(key)
    if raw is None:
        return None                        # 冷启动仍需一次性回源
    obj = json.loads(raw)
    if obj["expire_at"] > time.time():
        return obj["data"]                 # 逻辑未过期，直接用
    # 逻辑过期：抢锁异步刷新，当前请求拿旧值就走
    if r.set(f"lock:{key}", "1", nx=True, ex=10):
        threading.Thread(target=refresh, args=(key,), daemon=True).start()
    return obj["data"]

def refresh(key):
    try:
        data = db.query_hot(key)
        r.set(key, json.dumps({
            "data": data,
            "expire_at": time.time() + 3600,
        }))                                # 注意：不设 TTL
    finally:
        r.delete(f"lock:{key}")

场景1: 10000 个 key 的 TTL 都是 3600s → 1 小时后集体过期
场景2: Redis 实例宕机 → 全部请求穿透

import random

base_ttl = 3600                       # 基础 1 小时
jitter = random.randint(0, 600)       # 0-10 分钟抖动
r.setex(key, base_ttl + jitter, value)

请求 → 本地缓存 (Caffeine/Guava) → Redis → 数据库
L1 本地: TTL=60s，容量小，挡住热点重复读
L2 Redis: TTL=3600s，容量大
即使 Redis 短暂不可用，L1 仍能挡住部分请求

from circuitbreaker import circuit

@circuit(failure_threshold=5, recovery_timeout=30)
def get_data(key):
    data = r.get(key)
    if data:
        return json.loads(data)
    return db.query(key)

def get_data_fallback(key):
    return {"message": "服务繁忙，请稍后重试"}   # 兜底响应，保住可用性

def buggy_get(item_id):
    data = r.get(f"item:{item_id}")
    if data:
        return json.loads(data)
    db_data = db.query_item(item_id)
    if db_data:
        r.setex(f"item:{item_id}", 3600, json.dumps(db_data))
    return db_data