📚 课程目录第3章缓存空值回填方案

3. 缓存空值回填方案：实现原理、代码示例、TTL 设置策略

3.1 原理

缓存空值回填（Cache Null Object / Cache-Aside with Null）是在缓存穿透场景中，将数据库查询结果为 null 的 key 也写入缓存，但使用一个较短的 TTL（例如 30~60 秒）来标识“该 key 不存在”。后续对该 key 的请求直接命中缓存的空值对象，避免每次穿透到数据库。

核心收益：将“穿透”转化为“缓存命中”，保护数据库。空值对象体积极小（通常 1 字节），内存开销可忽略。

3.2 实现流程

应用查询缓存，key 未命中 → 查询数据库。
数据库返回结果：若为真实数据，回填缓存并设正常 TTL（如 3600s）；若结果不存在（即空值），回填标记值（如空字符串、特殊占位符 __NULL__），并设短 TTL（如 60s）。
应用返回响应（正常数据或“未找到”）。
后续请求命中缓存的空值，直接返回“未找到”，不再穿透数据库。

3.3 代码示例（Python + Redis）

import redis
import time

r = redis.Redis(host='localhost', port=6379, db=0)
NULL_PLACEHOLDER = b''  # 空值标记，使用空字节串降低存储开销
SHORT_TTL = 60          # 空值缓存时长（秒）
LONG_TTL = 600          # 正常数据缓存时长（秒）

def get_user(user_id: int) -> dict | None:
    key = f"user:{user_id}"
    # 1. 查询缓存
    cached = r.get(key)
    if cached is not None:
        if cached == NULL_PLACEHOLDER:
            return None  # 缓存中已标记为空，直接返回 None
        else:
            return json.loads(cached.decode('utf-8'))
    
    # 2. 缓存未命中，查询数据库（模拟）
    user_data = query_database(user_id)
    if user_data is not None:
        # 3a. 数据存在，回填并设置长 TTL
        r.setex(key, LONG_TTL, json.dumps(user_data).encode('utf-8'))
        return user_data
    else:
        # 3b. 数据不存在，回填空值标记并设置短 TTL
        r.setex(key, SHORT_TTL, NULL_PLACEHOLDER)
        return None

def query_database(user_id: int) -> dict | None:
    # 模拟数据库查询
    if user_id % 2 == 0:
        return {"id": user_id, "name": f"user_{user_id}"}
    else:
        return None  # 用户不存在

3.4 TTL 设置策略

场景	数据类型	推荐 TTL	说明
正常数据	真实对象	300~600s	按业务容忍延迟设置，可动态调整，避免频繁重建。
空值标记	空字节或特殊字符串	30~60s	短 TTL 确保空值标记自动过期，防止长期缓存“不存在”状态。（如数据新增后可以快速生效）
热点空值	频遭攻击的 key	尽量短（10~30s）	结合限流、布隆过滤器第一道防线；空值回填仅作为兜底。

  关键指标：空值回填的 TTL 不宜超过业务数据更新感知的 SLA。例如用户注册后需立即看到信息，则空值 TTL ≤ 注册后生效时间。若数据库变更后手动清除相关空值缓存，可用更短 TTL 或主动删除。

3.5 常见错误与避坑

空值标记与真实数据混淆：避免使用 -1、0、"" 等可能被误认为合法返回值的标记。建议使用统一常量（如 __NULL__）或空字节 b''。
TTL 设置过短：空值标记 TTL 若小于 10s，在高并发穿透下仍可能在极短时间内穿透数据库。建议最低 30s。
缓存雪崩叠加风险：大量空值标记同时过期可能导致突发回填压力。可加入随机抖动（±10s）打散过期时间。
遗漏串行化：高并发下多个线程同时检测到未命中并重复回填，可使用分布式锁（如 Redis SETNX）保护回填操作。

import random
def get_user_with_jitter(user_id):
    key = f"user:{user_id}"
    cached = r.get(key)
    if cached is not None:
        return process_cached(cached)
    # 加分布式锁进行回填
    lock_key = f"lock:{key}"
    if r.setnx(lock_key, 1):
        r.expire(lock_key, 5)
        try:
            data = query_database(user_id)
            if data:
                r.setex(key, LONG_TTL + random.randint(-60, 60), serialize(data))
            else:
                # 空值 TTL 也添加随机抖动
                ttl = SHORT_TTL + random.randint(-10, 10)
                r.setex(key, ttl, NULL_PLACEHOLDER)
            return data
        finally:
            r.delete(lock_key)
    else:
        time.sleep(0.01)
        return get_user_with_jitter(user_id)  # 重试

注意：空值回填无法完全防御非 key 型攻击（如分布式 ID 跳过布隆过滤器），需和布隆过滤器、限流组合使用。本方案适用于缓存层已经有完整 key 集合的场景。

3.6 性能与稳定性优化

内存占用：空值标记只存一个约定字节，每百万 key 仅增加约 1MB 内存（含 Redis 元数据）。
自动清理：短 TTL 到期 Redis 会自动淘汰，无需额外 scan 操作。
监控报警：监控缓存的空值标记比重。若超过整体缓存 key 的 30%，可能是攻击流量特征，触发限流策略。
结合异步更新：对于数据库新增记录，可发送异步消息清除对应空值缓存（使用 r.delete(key)），减少空变有时的等待窗口。