Mar 02 2023

Redis缓存穿透、击穿、雪崩问题终极解决方案：从布隆过滤器到多级缓存架构设计

作者： 绝缘体..   发布： 2023-03-02 15:01   分类： 未分类   阅读：   抢沙发  
 
更多

Redis缓存穿透、击穿、雪崩问题终极解决方案：从布隆过滤器到多级缓存架构设计

引言

在现代分布式系统中，Redis作为高性能的内存数据库，已成为缓存系统的核心组件。然而，在高并发场景下，缓存系统面临着三大经典问题：缓存穿透、缓存击穿和缓存雪崩。这些问题不仅会影响系统的性能，还可能导致整个服务的崩溃。本文将深入分析这三个问题的本质，并提供从布隆过滤器到多级缓存架构的完整解决方案。

缓存穿透问题详解与解决方案

什么是缓存穿透

缓存穿透是指查询一个根本不存在的数据，缓存层和存储层都不会命中，导致请求直接打到数据库上。这种情况通常发生在恶意攻击或数据异常的情况下，大量请求会直接访问数据库，造成数据库压力过大。

缓存穿透的危害

// 缓存穿透的典型场景示例
public class CachePenetrationDemo {
    private static final String CACHE_KEY_PREFIX = "user:";
    
    public User getUserById(Long id) {
        // 1. 先从缓存获取
        String cacheKey = CACHE_KEY_PREFIX + id;
        User user = redisTemplate.opsForValue().get(cacheKey);
        
        if (user != null) {
            return user;
        }
        
        // 2. 缓存未命中，查询数据库
        user = userDao.findById(id);
        
        if (user == null) {
            // 3. 数据库也未找到，此时需要处理空值
            // 这里存在缓存穿透风险
            return null;
        }
        
        // 4. 存储到缓存
        redisTemplate.opsForValue().set(cacheKey, user, 30, TimeUnit.MINUTES);
        return user;
    }
}

布隆过滤器解决方案

布隆过滤器是一种概率型数据结构，可以高效地判断一个元素是否存在于集合中。通过在缓存前增加布隆过滤器，可以有效防止缓存穿透问题。

@Component
public class BloomFilterCache {
    private final BloomFilter<String> bloomFilter;
    private final RedisTemplate<String, Object> redisTemplate;
    
    public BloomFilterCache(RedisTemplate<String, Object> redisTemplate) {
        this.redisTemplate = redisTemplate;
        // 初始化布隆过滤器，预计容量100万，误判率0.1%
        this.bloomFilter = BloomFilter.create(
            Funnels.stringFunnel(Charset.defaultCharset()),
            1000000,
            0.001
        );
    }
    
    /**
     * 检查key是否存在
     */
    public boolean exists(String key) {
        return bloomFilter.mightContain(key);
    }
    
    /**
     * 添加key到布隆过滤器
     */
    public void add(String key) {
        bloomFilter.put(key);
    }
    
    /**
     * 带布隆过滤器的缓存读取
     */
    public User getUserWithBloomFilter(Long userId) {
        String cacheKey = "user:" + userId;
        
        // 1. 先检查布隆过滤器
        if (!bloomFilter.mightContain(cacheKey)) {
            // 布隆过滤器判断不存在，直接返回null
            return null;
        }
        
        // 2. 缓存查询
        User user = (User) redisTemplate.opsForValue().get(cacheKey);
        if (user != null) {
            return user;
        }
        
        // 3. 缓存未命中，查询数据库
        user = userDao.findById(userId);
        
        if (user != null) {
            // 4. 数据库有数据，写入缓存和布隆过滤器
            redisTemplate.opsForValue().set(cacheKey, user, 30, TimeUnit.MINUTES);
            bloomFilter.put(cacheKey);
        } else {
            // 5. 数据库无数据，设置空值缓存，避免重复查询
            redisTemplate.opsForValue().set(cacheKey, "", 5, TimeUnit.MINUTES);
        }
        
        return user;
    }
}

空值缓存策略

对于查询不到的数据，可以将其空值也缓存起来，避免重复查询数据库：

@Service
public class UserService {
    private static final String CACHE_KEY_PREFIX = "user:";
    private static final int EMPTY_CACHE_TTL = 300; // 5分钟
    
    public User getUserById(Long id) {
        String cacheKey = CACHE_KEY_PREFIX + id;
        
        // 先从缓存获取
        User user = (User) redisTemplate.opsForValue().get(cacheKey);
        
        if (user == null) {
            // 缓存未命中，查询数据库
            user = userDao.findById(id);
            
            if (user == null) {
                // 数据库也未找到，缓存空值
                redisTemplate.opsForValue().set(cacheKey, "", EMPTY_CACHE_TTL, TimeUnit.SECONDS);
                return null;
            }
            
            // 数据库有数据，缓存到Redis
            redisTemplate.opsForValue().set(cacheKey, user, 30, TimeUnit.MINUTES);
        }
        
        return user;
    }
}

缓存击穿问题详解与解决方案

什么是缓存击穿

缓存击穿是指某个热点数据在缓存过期的瞬间，大量并发请求同时访问该数据，导致数据库压力骤增。与缓存穿透不同的是，缓存击穿中的数据是真实存在的，只是恰好在缓存失效的瞬间被大量访问。

缓存击穿的典型场景

// 缓存击穿的示例代码
public class CacheBreakdownDemo {
    private static final String HOT_KEY_PREFIX = "hot_data:";
    
    public String getHotData(String key) {
        String cacheKey = HOT_KEY_PREFIX + key;
        
        // 1. 从缓存获取
        String data = redisTemplate.opsForValue().get(cacheKey);
        
        if (data == null) {
            // 2. 缓存失效，需要从数据库获取
            // 多个线程同时执行到这里，造成数据库压力
            data = databaseService.getData(key);
            
            if (data != null) {
                redisTemplate.opsForValue().set(cacheKey, data, 60, TimeUnit.SECONDS);
            }
        }
        
        return data;
    }
}

分布式锁解决方案

使用分布式锁确保同一时间只有一个线程去查询数据库：

@Component
public class DistributedLockCache {
    private static final String LOCK_PREFIX = "lock:";
    private static final String CACHE_PREFIX = "cache:";
    
    public String getHotDataWithLock(String key) {
        String cacheKey = CACHE_PREFIX + key;
        String lockKey = LOCK_PREFIX + key;
        
        // 1. 先从缓存获取
        String data = redisTemplate.opsForValue().get(cacheKey);
        
        if (data != null) {
            return data;
        }
        
        // 2. 尝试获取分布式锁
        String lockValue = UUID.randomUUID().toString();
        Boolean acquired = redisTemplate.opsForValue()
            .setIfAbsent(lockKey, lockValue, 10, TimeUnit.SECONDS);
        
        try {
            if (acquired) {
                // 3. 获取锁成功，再次检查缓存
                data = redisTemplate.opsForValue().get(cacheKey);
                if (data != null) {
                    return data;
                }
                
                // 4. 缓存仍未命中，查询数据库
                data = databaseService.getData(key);
                
                if (data != null) {
                    // 5. 写入缓存
                    redisTemplate.opsForValue().set(cacheKey, data, 60, TimeUnit.SECONDS);
                } else {
                    // 6. 数据库无数据，设置空值缓存
                    redisTemplate.opsForValue().set(cacheKey, "", 300, TimeUnit.SECONDS);
                }
            } else {
                // 7. 获取锁失败，等待一段时间后重试
                Thread.sleep(100);
                return getHotDataWithLock(key);
            }
        } finally {
            // 8. 释放锁
            releaseLock(lockKey, lockValue);
        }
        
        return data;
    }
    
    private void releaseLock(String lockKey, String lockValue) {
        String script = 
            "if redis.call('get', KEYS[1]) == ARGV[1] then " +
            "return redis.call('del', KEYS[1]) " +
            "else return 0 end";
        
        redisTemplate.execute(new DefaultRedisScript<>(script, Long.class), 
                             Collections.singletonList(lockKey), lockValue);
    }
}

双重检查机制

结合缓存和分布式锁的双重检查机制：

@Component
public class DoubleCheckCache {
    private static final String CACHE_PREFIX = "double_check_cache:";
    private static final String LOCK_PREFIX = "double_check_lock:";
    
    public String getDataWithDoubleCheck(String key) {
        String cacheKey = CACHE_PREFIX + key;
        String lockKey = LOCK_PREFIX + key;
        
        // 第一次检查缓存
        String data = redisTemplate.opsForValue().get(cacheKey);
        if (data != null) {
            return data;
        }
        
        // 第二次检查分布式锁
        String lockValue = UUID.randomUUID().toString();
        Boolean acquired = redisTemplate.opsForValue()
            .setIfAbsent(lockKey, lockValue, 5, TimeUnit.SECONDS);
        
        if (acquired) {
            try {
                // 再次检查缓存（双重检查）
                data = redisTemplate.opsForValue().get(cacheKey);
                if (data != null) {
                    return data;
                }
                
                // 查询数据库
                data = databaseService.getData(key);
                
                if (data != null) {
                    redisTemplate.opsForValue().set(cacheKey, data, 60, TimeUnit.SECONDS);
                } else {
                    redisTemplate.opsForValue().set(cacheKey, "", 300, TimeUnit.SECONDS);
                }
                
                return data;
            } finally {
                releaseLock(lockKey, lockValue);
            }
        } else {
            // 等待后重试
            try {
                Thread.sleep(50);
                return getDataWithDoubleCheck(key);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return null;
            }
        }
    }
    
    private void releaseLock(String lockKey, String lockValue) {
        String script = "if redis.call('get', KEYS[1]) == ARGV[1] then " +
                       "return redis.call('del', KEYS[1]) else return 0 end";
        
        redisTemplate.execute(new DefaultRedisScript<>(script, Long.class),
                             Collections.singletonList(lockKey), lockValue);
    }
}

缓存雪崩问题详解与解决方案

什么是缓存雪崩

缓存雪崩是指在某一时刻大量缓存同时失效，导致所有请求都直接访问数据库，造成数据库压力剧增甚至宕机的现象。这通常发生在缓存集中过期或者缓存服务大规模故障时。

缓存雪崩的影响

// 缓存雪崩的模拟场景
public class CacheAvalancheDemo {
    @Scheduled(fixedRate = 1000)
    public void simulateCacheExpire() {
        // 模拟缓存集中过期的情况
        Set<String> keys = redisTemplate.keys("user:*");
        for (String key : keys) {
            // 设置随机过期时间，避免集中过期
            redisTemplate.expire(key, new Random().nextInt(3600), TimeUnit.SECONDS);
        }
    }
}

随机过期时间策略

为缓存设置随机的过期时间，避免集中过期：

@Component
public class RandomExpiryCache {
    private static final String CACHE_PREFIX = "random_expiry:";
    private static final int BASE_EXPIRY_SECONDS = 3600;
    private static final int RANDOM_RANGE = 300; // 5分钟随机范围
    
    public void setWithRandomExpiry(String key, Object value) {
        String cacheKey = CACHE_PREFIX + key;
        
        // 计算随机过期时间
        int randomExpiry = BASE_EXPIRY_SECONDS + new Random().nextInt(RANDOM_RANGE);
        
        redisTemplate.opsForValue().set(cacheKey, value, randomExpiry, TimeUnit.SECONDS);
    }
    
    public Object getWithRandomExpiry(String key) {
        String cacheKey = CACHE_PREFIX + key;
        return redisTemplate.opsForValue().get(cacheKey);
    }
}

多级缓存架构

构建多级缓存架构，降低单点故障影响：

@Component
public class MultiLevelCache {
    private static final String LOCAL_CACHE_KEY = "local_cache:";
    private static final String REDIS_CACHE_KEY = "redis_cache:";
    
    // 本地缓存（Caffeine）
    private final Cache<String, Object> localCache = Caffeine.newBuilder()
        .maximumSize(1000)
        .expireAfterWrite(10, TimeUnit.MINUTES)
        .build();
    
    // Redis缓存
    private final RedisTemplate<String, Object> redisTemplate;
    
    public Object getData(String key) {
        String localKey = LOCAL_CACHE_KEY + key;
        String redisKey = REDIS_CACHE_KEY + key;
        
        // 1. 先查本地缓存
        Object data = localCache.getIfPresent(localKey);
        if (data != null) {
            return data;
        }
        
        // 2. 本地缓存未命中，查Redis
        data = redisTemplate.opsForValue().get(redisKey);
        if (data != null) {
            // 3. Redis命中，放入本地缓存
            localCache.put(localKey, data);
            return data;
        }
        
        // 4. 两级缓存都未命中，查询数据库
        data = databaseService.getData(key);
        
        if (data != null) {
            // 5. 数据库有数据，写入两级缓存
            redisTemplate.opsForValue().set(redisKey, data, 60, TimeUnit.SECONDS);
            localCache.put(localKey, data);
        }
        
        return data;
    }
    
    public void putData(String key, Object value) {
        String localKey = LOCAL_CACHE_KEY + key;
        String redisKey = REDIS_CACHE_KEY + key;
        
        // 同时更新两级缓存
        redisTemplate.opsForValue().set(redisKey, value, 60, TimeUnit.SECONDS);
        localCache.put(localKey, value);
    }
}

热点数据预热策略

热点数据识别

通过监控和分析，识别出热点数据并进行预热：

@Component
public class HotDataPreheater {
    private static final String HOT_DATA_LIST_KEY = "hot_data_list";
    private static final String HOT_DATA_PREFIX = "hot_data:";
    
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    @Autowired
    private DatabaseService databaseService;
    
    /**
     * 预热热点数据
     */
    @Scheduled(cron = "0 0 2 * * ?") // 每天凌晨2点执行
    public void preheatHotData() {
        // 1. 获取热点数据列表（从监控系统获取）
        List<String> hotKeys = getHotDataList();
        
        // 2. 并发预热
        ExecutorService executor = Executors.newFixedThreadPool(10);
        
        for (String key : hotKeys) {
            executor.submit(() -> {
                try {
                    // 3. 预热数据
                    Object data = databaseService.getData(key);
                    if (data != null) {
                        String cacheKey = HOT_DATA_PREFIX + key;
                        redisTemplate.opsForValue().set(cacheKey, data, 60, TimeUnit.SECONDS);
                    }
                } catch (Exception e) {
                    log.error("预热数据失败: {}", key, e);
                }
            });
        }
        
        executor.shutdown();
        try {
            if (!executor.awaitTermination(30, TimeUnit.SECONDS)) {
                executor.shutdownNow();
            }
        } catch (InterruptedException e) {
            executor.shutdownNow();
        }
    }
    
    /**
     * 获取热点数据列表
     */
    private List<String> getHotDataList() {
        // 实际项目中可以从监控系统、日志分析等方式获取
        return Arrays.asList("user_1", "user_2", "product_1", "product_2");
    }
}

动态预热策略

根据实时访问模式动态调整预热策略：

@Component
public class DynamicPreheater {
    private static final String ACCESS_COUNT_KEY = "access_count:";
    private static final String PREHEAT_QUEUE_KEY = "preheat_queue";
    
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    @Autowired
    private DatabaseService databaseService;
    
    /**
     * 监控访问频率，动态触发预热
     */
    public void monitorAndPreheat(String key) {
        String accessCountKey = ACCESS_COUNT_KEY + key;
        
        // 统计访问次数
        Long count = redisTemplate.opsForValue().increment(accessCountKey);
        redisTemplate.expire(accessCountKey, 1, TimeUnit.HOURS);
        
        // 当访问次数达到阈值时触发预热
        if (count != null && count > 1000) {
            triggerPreheat(key);
        }
    }
    
    /**
     * 触发预热
     */
    private void triggerPreheat(String key) {
        // 将预热任务加入队列
        redisTemplate.opsForList().leftPush(PREHEAT_QUEUE_KEY, key);
        
        // 异步处理预热任务
        CompletableFuture.runAsync(() -> {
            try {
                Object data = databaseService.getData(key);
                if (data != null) {
                    String cacheKey = "hot_data:" + key;
                    redisTemplate.opsForValue().set(cacheKey, data, 60, TimeUnit.SECONDS);
                }
            } catch (Exception e) {
                log.error("预热失败: {}", key, e);
            }
        });
    }
}

完整的缓存系统设计方案

架构图设计

/**
 * 缓存系统架构设计
 */
public class CacheSystemArchitecture {
    
    /**
     * 多级缓存架构
     * Local Cache (Caffeine) -> Redis Cluster -> Database
     */
    public class MultiLevelCacheSystem {
        private final LocalCache localCache;
        private final RedisCluster redisCluster;
        private final Database database;
        
        public MultiLevelCacheSystem() {
            this.localCache = new LocalCache();
            this.redisCluster = new RedisCluster();
            this.database = new Database();
        }
        
        public Object get(String key) {
            // 1. 本地缓存查询
            Object result = localCache.get(key);
            if (result != null) {
                return result;
            }
            
            // 2. Redis查询
            result = redisCluster.get(key);
            if (result != null) {
                // 3. 本地缓存回填
                localCache.put(key, result);
                return result;
            }
            
            // 4. 数据库查询
            result = database.get(key);
            if (result != null) {
                // 5. 缓存写入
                redisCluster.put(key, result);
                localCache.put(key, result);
            }
            
            return result;
        }
    }
}

完整的解决方案实现

@Component
public class CompleteCacheSolution {
    private static final String CACHE_PREFIX = "complete_cache:";
    private static final String LOCK_PREFIX = "lock:";
    private static final String EMPTY_CACHE_VALUE = "EMPTY";
    
    private final RedisTemplate<String, Object> redisTemplate;
    private final Cache<String, Object> localCache = Caffeine.newBuilder()
        .maximumSize(1000)
        .expireAfterWrite(30, TimeUnit.MINUTES)
        .build();
    
    public CompleteCacheSolution(RedisTemplate<String, Object> redisTemplate) {
        this.redisTemplate = redisTemplate;
    }
    
    /**
     * 完整的缓存读取逻辑
     */
    public Object getComplete(String key) {
        String cacheKey = CACHE_PREFIX + key;
        String lockKey = LOCK_PREFIX + key;
        
        // 1. 本地缓存查询
        Object result = localCache.getIfPresent(cacheKey);
        if (result != null && !EMPTY_CACHE_VALUE.equals(result)) {
            return result;
        }
        
        // 2. Redis查询
        result = redisTemplate.opsForValue().get(cacheKey);
        if (result != null && !EMPTY_CACHE_VALUE.equals(result)) {
            // 3. 本地缓存回填
            localCache.put(cacheKey, result);
            return result;
        }
        
        // 4. 缓存未命中，使用分布式锁
        String lockValue = UUID.randomUUID().toString();
        Boolean acquired = redisTemplate.opsForValue()
            .setIfAbsent(lockKey, lockValue, 10, TimeUnit.SECONDS);
        
        try {
            if (acquired) {
                // 5. 再次检查缓存
                result = redisTemplate.opsForValue().get(cacheKey);
                if (result != null && !EMPTY_CACHE_VALUE.equals(result)) {
                    localCache.put(cacheKey, result);
                    return result;
                }
                
                // 6. 数据库查询
                result = databaseService.getData(key);
                
                if (result != null) {
                    // 7. 写入缓存
                    redisTemplate.opsForValue().set(cacheKey, result, 60, TimeUnit.SECONDS);
                    localCache.put(cacheKey, result);
                } else {
                    // 8. 空值缓存
                    redisTemplate.opsForValue().set(cacheKey, EMPTY_CACHE_VALUE, 300, TimeUnit.SECONDS);
                }
            } else {
                // 9. 等待后重试
                Thread.sleep(50);
                return getComplete(key);
            }
        } finally {
            // 10. 释放锁
            releaseLock(lockKey, lockValue);
        }
        
        return result;
    }
    
    /**
     * 缓存更新
     */
    public void update(String key, Object value) {
        String cacheKey = CACHE_PREFIX + key;
        
        // 更新所有层级缓存
        redisTemplate.opsForValue().set(cacheKey, value, 60, TimeUnit.SECONDS);
        localCache.put(cacheKey, value);
    }
    
    /**
     * 删除缓存
     */
    public void delete(String key) {
        String cacheKey = CACHE_PREFIX + key;
        
        redisTemplate.delete(cacheKey);
        localCache.invalidate(cacheKey);
    }
    
    private void releaseLock(String lockKey, String lockValue) {
        String script = "if redis.call('get', KEYS[1]) == ARGV[1] then " +
                       "return redis.call('del', KEYS[1]) else return 0 end";
        
        redisTemplate.execute(new DefaultRedisScript<>(script, Long.class),
                             Collections.singletonList(lockKey), lockValue);
    }
}

性能优化建议

缓存策略配置

# Redis缓存配置示例
spring:
  redis:
    host: localhost
    port: 6379
    timeout: 2000ms
    lettuce:
      pool:
        max-active: 20
        max-idle: 10
        min-idle: 5
        max-wait: -1ms
    cache:
      # 缓存过期时间配置
      expire-after-write: 60s
      maximum-size: 10000

监控和告警

@Component
public class CacheMonitor {
    private static final String METRIC_PREFIX = "cache.";
    
    @Autowired
    private MeterRegistry meterRegistry;
    
    public void recordCacheHit(String type) {
        Counter.builder(METRIC_PREFIX + "hit")
               .tag("type", type)
               .register(meterRegistry)
               .increment();
    }
    
    public void recordCacheMiss(String type) {
        Counter.builder(METRIC_PREFIX + "miss")
               .tag("type", type)
               .register(meterRegistry)
               .increment();
    }
    
    public void recordCacheError(String type) {
        Counter.builder(METRIC_PREFIX + "error")
               .tag("type", type)
               .register(meterRegistry)
               .increment();
    }
}

总结

Redis缓存系统的三大问题——缓存穿透、击穿、雪崩——在高并发场景下具有严重的危害性。通过本文介绍的解决方案，我们可以构建一个稳定可靠的缓存系统：

  1. 缓存穿透：通过布隆过滤器和空值缓存策略，有效防止无效请求冲击数据库
  2. 缓存击穿：使用分布式锁和双重检查机制，确保热点数据的并发安全
  3. 缓存雪崩：采用多级缓存架构和随机过期时间，避免集中失效

同时，配合热点数据预热、合理的缓存策略配置以及完善的监控告警机制，能够显著提升缓存系统的稳定性和性能。在实际应用中，需要根据具体的业务场景选择合适的解决方案，并持续优化缓存策略以适应业务发展需求。

通过这些技术手段的综合运用，我们可以构建出一个既高效又稳定的缓存系统，为业务的快速发展提供强有力的支持。

本文来自极简博客，作者：飞翔的鱼，转载请注明原文链接：Redis缓存穿透、击穿、雪崩问题终极解决方案：从布隆过滤器到多级缓存架构设计

打赏

本文固定链接: https://www.cxy163.net/archives/6478 | 绝缘体

该日志由 绝缘体.. 于 2023年03月02日 发表在 未分类 分类下， 你可以发表评论，并在保留原文地址及作者的情况下引用到你的网站或博客。
原创文章转载请注明: Redis缓存穿透、击穿、雪崩问题终极解决方案：从布隆过滤器到多级缓存架构设计 | 绝缘体
关键字: , , , ,
【上一篇】
【下一篇】

您可能还会对这些文章感兴趣！

  1. CVE-2022-0543
  2. docker进阶_dockerswarm
  3. docker进阶_docker-compose
  4. docker基础_网络模式
  5. 命令行清空memcache缓存的两种方法
  6. laravel 缓存系统的使用
  7. mysql 配置ssl ，java应用连接
  8. (转载)Go channel 的妙用
  9. MariaDB/MySQL配置文件my.cnf解读
  10. 如何在 Laravel 代码中正确地使用数据库事务
  11. Mybatis、Mybatis Generator、Mybatis-Plus、
  12. CentOS 上部署达梦数据库
  13. 使用Spring Boot Actuator实现应用监控
  14. Java 21虚拟线程性能优化实测：对比传统线程模型
  15. Spring Boot快速入门与实践
  16. 使用Spring Boot快速搭建Java应用

Redis缓存穿透、击穿、雪崩问题终极解决方案：从布隆过滤器到多级缓存架构设计：等您坐沙发呢！

发表评论


快捷键：Ctrl+Enter