缓存空对象

问题引入

在了解了什么是缓存穿透之后，我们需要找到一种方法来防护这种攻击。

核心问题：

• 查询不存在的数据时，无法将结果写入缓存
• 每次请求都会调用外部 API
• 恶意攻击者可以构造大量不存在的 key

解决思路：即使数据不存在，也把”空结果”缓存起来

这就是缓存空对象方案的核心思想。

什么是缓存空对象？

缓存空对象是指：当查询的数据不存在时，在缓存中存储一个特殊的标记，表示”该数据不存在”。

核心原理

正常情况：
- 数据存在 → 缓存真实数据
- 数据不存在 → 无法缓存 → 每次都查询外部 API

缓存空对象方案：
- 数据存在 → 缓存真实数据
- 数据不存在 → 缓存空标记 → 避免重复查询

与其他方案对比

工作原理

缓存空对象的流程

核心步骤：

1. 查询缓存
   ↓
2. 如果缓存中有特殊标记 "__NULL__"
   → 说明数据不存在，直接返回
   ↓
3. 如果缓存中有正常数据
   → 直接返回缓存数据
   ↓
4. 缓存未命中，调用外部 API
   ↓
5. 如果 API 返回 404 或空结果
   → 缓存空标记 "__NULL__"（较短过期时间）
   ↓
6. 如果 API 返回正常数据
   → 缓存真实数据（较长过期时间）

为什么需要特殊标记？

# 问题：如何区分"缓存未命中"和"数据不存在"？

场景 1：缓存未命中
- redis.get("weather:invalid_city") → None
- 需要调用外部 API

场景 2：数据不存在（已缓存空标记）
- redis.get("weather:invalid_city") → "__NULL__"
- 直接返回 None，不需要调用外部 API

# 解决方案：使用特殊标记区分
NULL_MARKER = "__NULL__"

if cached == NULL_MARKER:
    return None  # 数据不存在
elif cached:
    return cached  # 有缓存数据
else:
    # 缓存未命中，调用外部 API

实战：缓存空对象防护缓存穿透

基础实现

import redis

redis_client = redis.Redis(host='localhost', port=6379, db=0)
NULL_MARKER = '__NULL__'

def get_weather(city):
    """
    查询天气信息，带缓存空对象防护
    """
    cache_key = f"weather:{city}"

    # 1. 先查缓存
    cached = redis_client.get(cache_key)

    # 2. 缓存中有特殊标记，表示数据不存在
    if cached == NULL_MARKER:
        return None

    # 3. 缓存中有正常数据
    if cached:
        return cached

    # 4. 缓存未命中，调用外部 API
    response = external_api.get_weather(city)

    if response.status == 404:
        # 5. 城市不存在，也缓存一个空标记
        redis_client.setex(cache_key, 60, NULL_MARKER)  # 空值缓存 1 分钟
        return None

    # 6. 数据存在，正常缓存
    redis_client.setex(cache_key, 3600, response.data)
    return response.data

使用装饰器的实现

import redis
from functools import wraps

NULL_MARKER = '__NULL__'

def cache_protect(null_ttl=60, normal_ttl=3600):
    """
    缓存穿透防护装饰器
    - null_ttl: 空值缓存时间（秒）
    - normal_ttl: 正常数据缓存时间（秒）
    """
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # 生成缓存 key
            cache_key = f"{func.__name__}:{args}:{kwargs}"

            # 查缓存
            cached = redis.get(cache_key)

            # 空值标记，直接返回
            if cached == NULL_MARKER:
                return None

            if cached:
                return cached

            # 调用外部 API
            try:
                result = func(*args, **kwargs)
            except Exception as e:
                # 查询失败，也缓存空值，防止持续穿透
                redis.setex(cache_key, null_ttl, NULL_MARKER)
                raise

            if result is None:
                # 结果为空，缓存空值
                redis.setex(cache_key, null_ttl, NULL_MARKER)
            else:
                # 结果正常，缓存数据
                redis.setex(cache_key, normal_ttl, result)

            return result

        return wrapper
    return decorator

# 使用示例
@cache_protect(null_ttl=60, normal_ttl=3600)
def get_weather(city):
    # 调用外部天气 API
    return external_api.get_weather(city)

完整的 API 实现

import redis
import json

redis_client = redis.Redis(host='localhost', port=6379, db=0)
NULL_MARKER = '__NULL__'

def get_weather(city):
    """
    查询天气信息，带缓存穿透防护
    """
    cache_key = f"weather:{city}"

    # 1. 查缓存
    cached = redis_client.get(cache_key)

    # 2. 空值标记，直接返回
    if cached == NULL_MARKER:
        return None

    if cached:
        return json.loads(cached)

    # 3. 调用外部 API
    try:
        response = external_api.get_weather(city)

        if response.status == 404:
            # 4. 城市不存在，缓存空值（60 秒过期）
            redis_client.setex(cache_key, 60, NULL_MARKER)
            return None

        # 5. 数据存在，正常缓存（1 小时过期）
        redis_client.setex(cache_key, 3600, json.dumps(response.data))
        return response.data

    except Exception as e:
        # API 调用失败，也缓存空值，防止持续穿透
        redis_client.setex(cache_key, 60, NULL_MARKER)
        raise

# 使用示例
@app.route('/api/weather')
def weather_api():
    city = request.args.get('city')

    # 参数校验（第一道防线）
    if not city or len(city) < 2:
        return {'error': '无效的城市名'}, 400

    try:
        data = get_weather(city)

        if not data:
            return {'error': '城市不存在'}, 404

        return {'data': data}

    except Exception as e:
        return {'error': str(e)}, 500

效果验证

防护前后对比

实施缓存空对象方案后，系统指标明显改善：

攻击拦截统计

防护期间统计（24 小时）：

拦截的恶意请求：
- 不存在的城市：50,000+ 次
- 无效的城市名：30,000+ 次
- 伪造的参数：20,000+ 次

外部 API 保护：
- 避免的 API 调用：100,000+ 次
- 节省的调用额度：约 60%

当前技术架构

课后练习

练习 1

缓存空对象方案中，为什么空值的过期时间要设置得比较短？

练习 2

缓存空对象方案和布隆过滤器方案有什么区别？分别适用于什么场景？

练习 3

请实现一个带有缓存空对象防护的用户查询函数，要求：

1. 使用装饰器实现
2. 空值缓存 60 秒，正常数据缓存 1 小时
3. 包含错误处理

import redis
import json
from functools import wraps

redis_client = redis.Redis(host='localhost', port=6379, db=0)
NULL_MARKER = '__NULL__'

def cache_protect(null_ttl=60, normal_ttl=3600):
    """
    缓存穿透防护装饰器

    Args:
        null_ttl: 空值缓存时间（秒）
        normal_ttl: 正常数据缓存时间（秒）
    """
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # 生成缓存 key
            cache_key = f"{func.__name__}:{args}:{kwargs}"

            # 查缓存
            try:
                cached = redis_client.get(cache_key)
            except redis.ConnectionError:
                # Redis 连接失败，直接调用函数
                return func(*args, **kwargs)

            # 空值标记，直接返回
            if cached == NULL_MARKER:
                return None

            # 有缓存数据
            if cached:
                try:
                    return json.loads(cached)
                except json.JSONDecodeError:
                    return cached

            # 缓存未命中，调用原函数
            try:
                result = func(*args, **kwargs)
            except Exception as e:
                # 查询失败，也缓存空值，防止持续穿透
                try:
                    redis_client.setex(cache_key, null_ttl, NULL_MARKER)
                except redis.ConnectionError:
                    pass
                raise

            # 缓存结果
            try:
                if result is None:
                    # 结果为空，缓存空值
                    redis_client.setex(cache_key, null_ttl, NULL_MARKER)
                else:
                    # 结果正常，缓存数据
                    if isinstance(result, (dict, list)):
                        redis_client.setex(cache_key, normal_ttl, json.dumps(result))
                    else:
                        redis_client.setex(cache_key, normal_ttl, result)
            except redis.ConnectionError:
                pass

            return result

        return wrapper
    return decorator

# 使用示例
@cache_protect(null_ttl=60, normal_ttl=3600)
def get_user(user_id):
    """
    查询用户信息，带缓存穿透防护

    Args:
        user_id: 用户 ID

    Returns:
        用户信息字典，如果不存在则返回 None
    """
    # 参数校验
    if not user_id or not isinstance(user_id, int):
        raise ValueError("无效的用户 ID")

    # 调用外部 API 或数据库
    try:
        response = external_api.get_user(user_id)

        if response.status == 404:
            return None

        return response.data

    except Exception as e:
        # API 调用失败
        raise Exception(f"查询用户失败: {str(e)}")

# 使用示例
@app.route('/api/users/<int:user_id>')
def user_api(user_id):
    try:
        user = get_user(user_id)

        if not user:
            return {'error': '用户不存在'}, 404

        return {'data': user}

    except ValueError as e:
        return {'error': str(e)}, 400
    except Exception as e:
        return {'error': '服务器错误'}, 500