Redis缓存穿透的成因与诊断

缓存穿透指大量请求查询一个在缓存和数据库中都不存在的key，请求绕过缓存直达数据库。典型场景：恶意爬虫用不存在的ID遍历接口、业务上线前预热遗漏的空值key、参数拼接错误导致的无效查询。穿透的杀伤力在于每个请求都绕过缓存层，在高并发下数据库连接池迅速耗尽。

诊断方法：监控Redis的keyspace_misses指标与数据库QPS，当misses指标突增且数据库QPS同步飙升时，大概率是穿透问题。另一个特征是慢查询日志中出现大量SELECT … WHERE id = ?返回0行的记录。

布隆过滤器防穿透实战

布隆过滤器（Bloom Filter）是一种空间高效的概率型数据结构，可以判断一个元素”一定不存在”或”可能存在”。在缓存前置布隆过滤器，查询请求先过布隆过滤器——如果判断key不存在，直接返回null不查数据库；如果判断可能存在，再走缓存→数据库的常规路径。

# Redis布隆过滤器实现（需RedisBloom模块）
// 1. 初始化：将所有合法ID加载到布隆过滤器
const redis = require('redis');
const client = redis.createClient();

async function initBloomFilter() {
    const allUserIds = await db.query('SELECT id FROM users');
    const pipeline = client.pipeline();

    for (const row of allUserIds) {
        // BF.ADD添加元素到布隆过滤器
        pipeline.call('BF.ADD', 'user:bloom', row.id.toString());
    }
    await pipeline.exec();
}

// 2. 查询时先检查布隆过滤器
async function getUser(userId) {
    // 先检查布隆过滤器
    const exists = await client.call('BF.EXISTS', 'user:bloom', userId.toString());
    if (!exists) {
        return null;  // 一定不存在，直接返回
    }

    // 可能存在，走缓存→数据库路径
    const cacheKey = `user:${userId}`;
    let data = await client.get(cacheKey);

    if (!data) {
        data = await db.query('SELECT * FROM users WHERE id = ?', [userId]);
        if (data) {
            await client.set(cacheKey, JSON.stringify(data), 'EX', 3600);
        } else {
            // 防止穿透：缓存空值，短TTL
            await client.set(cacheKey, 'NULL', 'EX', 60);
        }
    } else if (data === 'NULL') {
        return null;
    }

    return data;
}

布隆过滤器存在误判率（false positive），误判时请求仍会穿透到数据库，但概率可控。通过调整BF.RESERVE的error_rate和capacity参数，误判率可降到0.01%以下。布隆过滤器不支持删除，如果业务需要删除元素，改用Counting Bloom Filter或在元素变更时重建过滤器。

缓存空值方案与短TTL策略

不依赖RedisBloom模块的轻量替代方案是缓存空值：当数据库查询返回空结果时，在Redis中写入一个特殊标记值（如”NULL”），设置短TTL（30-120秒）。后续相同key的请求命中缓存中的空值标记，直接返回不查数据库。

// 缓存空值方案
public <T> T getWithNullCache(String key, Class<T> type, Supplier<T> dbQuery) {
    String value = redis.get(key);

    if (value != null) {
        if ("NULL".equals(value)) {
            return null;  // 命中空值缓存
        }
        return JSON.parseObject(value, type);
    }

    // 缓存未命中，查数据库
    T data = dbQuery.get();

    if (data != null) {
        redis.setex(key, 3600, JSON.toJSONString(data));  // 正常数据1小时
    } else {
        redis.setex(key, 60, "NULL");  // 空值缓存60秒
    }

    return data;
}

空值缓存的TTL不能太长——如果数据库中新增了该key的数据，空值缓存会导致一段时间内用户看不到新数据。60-120秒是较为合理的折中值，业务对实时性要求高的场景可缩短到15秒。配合消息队列监听数据变更事件主动清除空值缓存，可进一步缩短不一致窗口。

缓存雪崩的成因与防护

缓存雪崩指大量key在同一时刻集中过期，或Redis集群整体不可用，导致请求全部涌向数据库。与穿透不同，雪崩的key是真实存在的，只是同时失效了。最常见的原因是批量设置缓存时使用了相同的TTL——所有key在TTL到期那一秒同时失效。

TTL随机偏移与多级缓存

防止key集中过期的核心手段是TTL加随机偏移。设置缓存时不使用固定TTL，而是在基准TTL基础上加一个随机值（基准TTL的10%-30%），打散过期时间点。

import random
import redis

r = redis.Redis()

def set_cache_with_jitter(key, value, base_ttl=3600):
    """设置缓存并加入TTL随机偏移"""
    jitter = random.randint(0, int(base_ttl * 0.3))  # 0~30%的随机偏移
    ttl = base_ttl + jitter
    r.setex(key, ttl, value)

# 批量预热时效果显著
# 10000个key，base_ttl=3600，偏移后过期时间分布在3600~4680秒
# 避免第3600秒同时过期10800个key

多级缓存是另一道防线。L1用进程内缓存（Caffeine/Guava Cache），L2用Redis。请求先查L1，miss再查L2，miss再查数据库。L1缓存没有网络开销，能扛住Redis短时间不可用的场景。L1的TTL设置比L2短（L1: 60s, L2: 3600s），L1过期后回源L2而非数据库。

// Java多级缓存实现（Caffeine + Redis）
@Configuration
public class MultiLevelCacheConfig {

    @Bean
    public Cache<String, String> localCache() {
        return Caffeine.newBuilder()
            .maximumSize(10000)
            .expireAfterWrite(60, TimeUnit.SECONDS)    // L1: 60秒
            .recordStats()
            .build();
    }
}

@Service
public class CacheService {

    @Autowired
    private Cache<String, String> localCache;
    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    public String get(String key) {
        // L1: Caffeine本地缓存
        String value = localCache.getIfPresent(key);
        if (value != null) {
            return value;
        }

        // L2: Redis分布式缓存
        value = redisTemplate.opsForValue().get(key);
        if (value != null) {
            localCache.put(key, value);  // 回填L1
            return value;
        }

        return null;  // 两级缓存均miss
    }

    public void put(String key, String value, long redisTtlSeconds) {
        localCache.put(key, value);
        redisTemplate.opsForValue().set(key, value, redisTtlSeconds, TimeUnit.SECONDS);
    }
}

Redis集群故障的熔断保护

当Redis集群整体不可用时（网络分区、主从切换、OOM），应用不应无限等待或反复重试——这会让线程池耗尽，拖垮整个服务。熔断器（Circuit Breaker）在检测到Redis连续失败N次后进入断路状态，后续请求跳过Redis直接查数据库或返回降级数据。一段时间后进入半开状态，放行少量请求探测Redis是否恢复，恢复则闭合断路器。

// Resilience4j熔断器配置
CircuitBreakerConfig config = CircuitBreakerConfig.custom()
    .failureRateThreshold(50)            // 失败率超50%触发断路
    .waitDurationInOpenState(Duration.ofSeconds(30))  // 断路30秒
    .permittedNumberOfCallsInHalfOpenState(5)         // 半开状态放行5次
    .slidingWindowType(SlidingWindowType.COUNTING)
    .slidingWindowSize(20)               // 滑动窗口20次调用
    .build();

CircuitBreaker circuitBreaker = CircuitBreaker.of("redis", config);

public String getWithCircuitBreaker(String key) {
    return CircuitBreaker.decorateSupplier(circuitBreaker, () -> {
        String value = redisTemplate.opsForValue().get(key);
        if (value == null) throw new RedisKeyNotFoundException();  // key不存在不算失败
        return value;
    }).get();
}

// 熔断后的降级逻辑
public String getWithFallback(String key) {
    Supplier<String> supplier = CircuitBreaker.decorateSupplier(circuitBreaker, () ->
        redisTemplate.opsForValue().get(key)
    );
    Supplier<String> fallback = SupplierUtils.decorateSupplierWithFallback(
        supplier, circuitBreaker, e -> {
            log.warn("Redis circuit breaker open, falling back to DB");
            return dbQuery(key);  // 降级查数据库
        }
    );
    return fallback.get();
}

热点key的识别与本地缓存加固

雪崩场景下还有一种容易被忽视的子类型——热点key失效。某个key的QPS占整体10%以上（如热门商品详情、热搜词条），该key过期的瞬间大量请求同时回源数据库。常规TTL偏移无法解决，因为偏移值相对于大流量只是几秒的延迟。

热点key的防护手段：一是在应用层做本地缓存，热点key永远有一份进程内副本；二是热点key设置不设过期时间（TTL=-1），通过数据变更事件主动更新缓存。结合Redis的CLIENT LIST和MONITOR命令可以识别热点key——连接数最多的client或出现频率最高的key即为热点。

# Redis热点key分析脚本
import redis
from collections import Counter

r = redis.Redis()

# 方法1: 使用redis-cli --hot-keys（需要maxmemory-policy为LFU）
# redis-cli --hot-keys

# 方法2: 短时间采样MONITOR输出统计key频率
def find_hot_keys(duration_seconds=10):
    """采样MONITOR输出，统计key访问频率"""
    key_counter = Counter()
    start = time.time()

    with r.monitor() as monitor:
        for command in monitor:
            if time.time() - start > duration_seconds:
                break
            if len(command['args']) > 1 and command['args'][0] in ('GET', 'HGET', 'HGETALL'):
                key = command['args'][1]
                key_counter[key] += 1

    return key_counter.most_common(20)