# 连接本地 Redis（默认端口 6379）
redis-cli

# 故意查询一个不存在的 key，确认返回 null
GET user:999999
# (nil)

# 现在想象有 10000 个并发请求查询同一个不存在的 ID
# 每个请求都会穿透缓存，直接打到 MySQL

# 100 并发，发送 10000 次请求，查询不存在的用户
wrk -t10 -c100 -d10s http://localhost:8080/api/user/999999

from pybloom_live import ScalableBloomFilter
import redis

# 初始化布隆过滤器，误判率 0.1%
bloom = ScalableBloomFilter(initial_capacity=1000000, error_rate=0.001)

# 预热：将所有已存在的用户 ID 加入布隆过滤器
def warmup_bloom():
    # 从数据库加载所有有效用户 ID（实际应分页加载避免 OOM）
    valid_ids = get_all_valid_user_ids_from_db()
    for uid in valid_ids:
        bloom.add(uid)
    print(f"布隆过滤器预热完成，已加载 {len(valid_ids)} 个 ID")

# 查询接口的前置检查
def get_user(user_id):
    # 第一步：布隆过滤器检查（一定不存在 → 直接返回 404）
    if user_id not in bloom:
        print(f"[拦截] 用户 {user_id} 一定不存在，直接返回 404")
        return None

    # 第二步：查缓存（可能存在 → 继续查）
    cache_key = f"user:{user_id}"
    cached = redis_client.get(cache_key)
    if cached:
        return cached

    # 第三步：查数据库，结果写回缓存（含空值）
    user = db_query_user(user_id)
    if user:
        redis_client.setex(cache_key, 3600, user)
    else:
        # 空值缓存，短 TTL 防止误判过期后数据才入库的情况
        redis_client.setex(cache_key, 60, "NULL")
        print(f"[空值缓存] user:{user_id} 不存在，缓存 60 秒")
    return user

# 在 DAO 层统一处理：查到空结果也缓存
def query_with_cache(key, query_fn, ttl=3600, null_ttl=60):
    val = redis_client.get(key)
    if val is not None:
        return None if val == "NULL" else val
    
    result = query_fn()
    if result is None:
        redis_client.setex(key, null_ttl, "NULL")
    else:
        redis_client.setex(key, ttl, result)
    return result

# 1. 设置一个热点 key，TTL 为 5 秒
redis-cli SET hot_article:1 "热门文章内容" EX 5

# 2. 等待 5 秒让 key 过期（用 watch 观察）
watch -n 0.1 'redis-cli GET hot_article:1'

# 3. 在 key 过期的瞬间，用压测工具发起高并发请求
date && wrk -t20 -c500 -d2s http://localhost:8080/api/article/1

# ❌ 有击穿风险的写法（最常见错误）
def get_article(article_id):
    cache_key = f"article:{article_id}"
    cached = redis_client.get(cache_key)
    if cached:
        return cached
    # 这里没有并发控制！100 个请求同时发现缓存为空，
    # 会同时执行下面的数据库查询，产生 100 倍数据库压力！
    article = db_query_article(article_id)
    redis_client.setex(cache_key, 300, article)
    return article

import time
import redis

def get_article_with_lock(article_id, timeout=5):
    cache_key = f"article:{article_id}"
    lock_key = f"lock:article:{article_id}"
    
    # 先查缓存（快速路径）
    cached = redis_client.get(cache_key)
    if cached:
        return cached
    
    # 尝试获取分布式锁（SET NX EX，原子操作）
    acquired = redis_client.set(lock_key, "1", nx=True, ex=timeout)
    if acquired:
        try:
            # 获取到锁的线程：查数据库 → 写缓存
            article = db_query_article(article_id)
            redis_client.setex(cache_key, 300, article)
            return article
        finally:
            redis_client.delete(lock_key)
    else:
        # 没获取到锁的线程：等待后重试或返回旧值（如果有）
        time.sleep(0.05)  # 短暂休眠避免忙等（可优化为信号量）
        # 重试读缓存（此时持有锁的线程应该已经写入了）
        return redis_client.get(cache_key) or db_query_article(article_id)

import threading
import json

def get_article_logical_expiry(article_id):
    cache_key = f"article:{article_id}"
    cached = redis_client.get(cache_key)
    
    if not cached:
        # 缓存完全不存在时，同步加载（只会在首次触发）
        article = db_query_article(article_id)
        # 存储格式：{"data": ..., "expire_at": <unix_timestamp>}
        payload = {"data": article, "expire_at": time.time() + 300}
        redis_client.set(cache_key, json.dumps(payload))
        return article
    
    payload = json.loads(cached)
    
    if time.time() < payload["expire_at"]:
        # 逻辑未过期，直接返回数据（注意：即使物理 key 永不过期也要判断逻辑时间）
        return payload["data"]
    else:
        # 逻辑已过期，但先返回旧值，异步刷新缓存（用户体验优先）
        threading.Thread(target=async_refresh_article, args=(article_id, cache_key), daemon=True).start()
        return payload["data"]

def async_refresh_article(article_id, cache_key):
    article = db_query_article(article_id)
    payload = {"data": article, "expire_at": time.time() + 300}
    redis_client.set(cache_key, json.dumps(payload))
    print(f"[异步刷新] article:{article_id} 缓存已更新")

# 模拟批量设置缓存（注意：所有 key 的 TTL 都是 300 秒）
redis-cli MSET item:1 "data1" item:2 "data2" item:3 "data3" ... item:10000 "data10000"

# 批量设置过期时间（问题所在：所有 key 在同一时间过期）
for i in $(seq 1 10000); do
    redis-cli EXPIRE item:$i 300
done

# 300 秒后，所有 key 同时过期
# 此时如果有 10000 QPS 的请求到来，全部穿透缓存打到数据库
date && wrk -t50 -c2000 -d5s http://localhost:8080/api/item/{1..10000}

import random

def set_cache_with_jitter(key, value, base_ttl=300):
    """
    在基础 TTL 上增加 ±20% 的随机抖动
    例：base_ttl=300 → 实际 TTL 在 240~360 秒之间随机分布
    这样即使 10000 个 key 同时设置，也会在 2 分钟内分批过期，而非同时过期。
    """
    jitter = int(base_ttl * 0.2 * (2 * random.random() - 1))  # ±20%
    actual_ttl = base_ttl + jitter
    redis_client.setex(key, actual_ttl, value)
    # 调试日志：print(f"设置 {key}, TTL={actual_ttl}s (base={base_ttl}s, jitter={jitter})")

def warmup_hot_keys():
    """
    服务启动时预热热点 key，避免冷启动时缓存空窗期。
    同时给预热 key 设置随机 TTL，防止后续集体过期。
    """
    hot_keys = get_hot_keys_from_config()  # 从配置或数据库中读取需要预热的 key 列表
    
    for key_template in hot_keys:
        # 查询数据（带重试机制）
        data = db_query_with_retry(key_template, max_retries=3)
        if data:
            base_ttl = get_ttl_for_key_type(key_template)
            set_cache_with_jitter(key_template, data, base_ttl)
            print(f"[预热] {key_template} 已缓存")
        else:
            print(f"[预热跳过] {key_template} 数据不存在")
    
    print(f"[预热完成] 共加载 {len(hot_keys)} 个热点 key")

def db_query_with_retry(key, max_retries=3):
    """带重试的数据库查询，防止预热阶段数据库瞬时压力过大"""
    for attempt in range(max_retries):
        try:
            return db_query(key)
        except Exception as e:
            if attempt == max_retries - 1:
                print(f"[预热失败] {key} 查询失败: {e}")
                return None
            time.sleep(0.1 * (2 ** attempt))  # 指数退避：0.1s, 0.2s, 0.4s

from functools import lru_cache
import threading

# 本地缓存（进程内，容量有限但速度极快）
@lru_cache(maxsize=10000)
def local_cache_get(key):
    return None  # lru_cache 本身不提供 TTL，需配合其他方案（如 cachetools）

def get_data_multi_level(key, db_query_fn, redis_ttl=300, local_ttl=60):
    """
    三级缓存策略：本地缓存 → Redis → 数据库
    - L1: 进程内 LRU 缓存，极快但容量有限，适合极热点数据（单机 10000 条）
    - L2: Redis 分布式缓存，容量大，适合大部分热数据（集群 1000 万条）
    - L3: 数据库，持久化数据源，容量最大但最慢（亿级数据）
    
    降级逻辑：
    - Redis 宕机 → 降级为 L1 + L3（本地缓存兜底）
    - 本地缓存命中 → 不经过 Redis 和 DB（最快路径）
    """
    # L1: 本地缓存检查（进程内，微秒级）
    local_val = local_cache_get(key)
    if local_val:
        return local_val
    
    # L2: Redis 缓存检查（网络请求，毫秒级）
    try:
        redis_val = redis_client.get(key)
        if redis_val:
            # 写入本地缓存（双写一致性：L2 命中时也更新 L1）
            local_cache_get.cache_clear()  # 实际应用中应使用更精细的更新策略（如 cachetools）
            return redis_val
    except redis.ConnectionError as e:
        print(f"[降级] Redis 不可用，跳过 L2 缓存: {e}")
        # Redis 宕机时的降级逻辑：直接查数据库，同时缓存到本地（跳过 L2）
        data = db_query_fn()
        if data:
            # 本地缓存兜底，但 TTL 更短（防止本地缓存过期导致后续大量请求打到 DB）
            _set_local_with_ttl(key, data, local_ttl)  # 需要自定义实现
        return data
    
    # L3: 数据库查询（磁盘 I/O，最慢）
    data = db_query_fn()
    if data:
        redis_client.setex(key, redis_ttl, data)
        _set_local_with_ttl(key, data, local_ttl)
    return data

### 5.2 验证缓存击穿防护生效（并发压测）
```bash
# 1. 清空缓存中某个热点 key（模拟过期）
redis-cli DEL hot_article:1

# 2. 用 wrk 发起高并发请求（500 并发，2 秒持续）
wrk -t20 -c500 -d2s http://localhost:8080/api/article/1

# 3. 同时监控 MySQL 连接数和慢查询
echo "--- MySQL 当前连接数 ---"
mysql -u root -e "SHOW STATUS LIKE 'Threads_connected';"

# 预期结果：
# - 加了互斥锁：Threads_connected 基本不变（只有 1~2 个查库线程）
# - 没加锁：Threads_connected 飙升到几百，出现 Too many connections 错误

import redis
import time

# 创建 100 个 key，使用抖动后的 TTL
def verify_ttl_jitter():
    r = redis.Redis()
    base_ttl = 300
    for i in range(100):
        r.setex(f"test_key:{i}", 300, f"value_{i}")
    
    # 检查每个 key 的实际剩余 TTL（应该分布在 240~360 秒之间）
    ttls = []
    for i in range(100):
        ttl = r.ttl(f"test_key:{i}")
        ttls.append(ttl)
    
    print(f"TTL 分布统计:")
    print(f"  最小值: {min(ttls)}s")
    print(f"  最大值: {max(ttls)}s")
    print(f"  平均值: {sum(ttls)/len(ttls):.1f}s")
    print(f"  范围: {max(ttls) - min(ttls)}s")
    print(f"  预期范围: {int(base_ttl * 0.8)}s ~ {int(base_ttl * 1.2)}s")
    assert min(ttls) >= int(base_ttl * 0.8), "TTL 抖动范围不符合预期！"
    assert max(ttls) <= int(base_ttl * 1.2), "TTL 抖动范围不符合预期！"
    print("✅ TTL 抖动验证通过！")

verify_ttl_jitter()

# Prometheus + Alertmanager 规则示例（prometheus-rules.yml）
groups:
  - name: cache_alerts
    rules:
      # 规则 1：缓存命中率低于 70% 持续 5 分钟 → 告警（说明缓存命中率骤降，可能有大面积失效）
      - alert: CacheHitRateLow
        expr: redis_cache_hit_rate < 0.70
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "缓存命中率低于 70%，可能存在缓存失效问题"
          description: "当前命中率: {{ $value }}%，请立即检查缓存 key 的过期设置"
      
      # 规则 2：Redis 实例不可用 → 紧急告警（说明整个缓存层挂了）
      - alert: RedisDown
        expr: redis_up == 0
        for: 30s
        labels:
          severity: critical
        annotations:
          summary: "Redis 实例 {{ $labels.instance }} 不可用！"
          description: "Redis 已宕机 {{ $value }} 秒，服务已降级到数据库查询，请立即处理！"
      
      # 规则 3：数据库连接数超过 80% → 预警（可能是缓存穿透导致的连锁反应）
      - alert: DbConnectionHigh
        expr: mysql_threads_connected / mysql_max_connections > 0.80
        for: 2m
        labels:
          severity: warning
        annotations:
          summary: "数据库连接数超过 80%"
          description: "当前连接: {{ $value }}%，可能是缓存失效导致请求全部打到数据库"

---

## 七、总结：一套完整的缓存防护架构长什么样？

把上面的方案组合起来，一个生产级的缓存查询流程应该是这样的：