Nov 07 2024

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

作者： 绝缘体..   发布： 2024-11-07 09:47   分类： 未分类   阅读：   抢沙发  
 
更多

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

在现代高并发的互联网应用中，Redis作为高性能的内存数据库，承担着缓存核心数据、减轻数据库压力的重要职责。然而，随着业务规模的不断扩大，缓存系统面临着三个核心挑战：缓存穿透、缓存击穿和缓存雪崩。这些问题不仅会影响系统性能，严重时甚至会导致整个服务不可用。

本文将深入分析这三种缓存问题的本质，提供包括布隆过滤器、热点数据预热、分布式锁、多级缓存等完整的解决方案，帮助企业构建高可用的缓存架构。

什么是缓存穿透、击穿、雪崩

缓存穿透（Cache Penetration）

缓存穿透是指查询一个根本不存在的数据，由于缓存层没有命中，请求会穿透到数据库层。如果数据库中也不存在该数据，则不会写入缓存，导致每次请求都会直接访问数据库。在高并发场景下，这种恶意攻击或大量无效请求会严重拖垮数据库。

缓存击穿（Cache Breakdown）

缓存击穿是指某个热点key在缓存中过期的瞬间，大量并发请求同时访问该key，导致所有请求都穿透到数据库，造成数据库瞬间压力激增。

缓存雪崩（Cache Avalanche）

缓存雪崩是指大量缓存key在同一时间失效，或者Redis服务宕机，导致大量请求直接访问数据库，引发数据库压力过大甚至宕机，最终导致整个系统瘫痪。

缓存穿透解决方案

1. 布隆过滤器（Bloom Filter）

布隆过滤器是一种空间效率极高的概率型数据结构，用于判断一个元素是否存在于集合中。它能够快速判断某个key肯定不存在，从而避免无效请求穿透到数据库。

布隆过滤器原理

布隆过滤器由一个位数组和多个哈希函数组成。当添加元素时，通过多个哈希函数计算出多个位置，并将这些位置设为1。查询时，如果所有哈希函数计算出的位置都为1，则认为元素可能存在；如果有一个位置为0，则元素肯定不存在。

布隆过滤器实现

import java.util.BitSet;

public class BloomFilter {
    private BitSet bitSet;
    private int bitSize;
    private int[] hashSeeds;
    
    public BloomFilter(int capacity, double errorRate) {
        // 计算位数组大小
        this.bitSize = (int) Math.ceil(capacity * Math.log(errorRate) / Math.log(0.5) / Math.log(2));
        this.bitSet = new BitSet(bitSize);
        
        // 计算哈希函数个数
        int hashCount = (int) Math.ceil(Math.log(2) * bitSize / capacity);
        this.hashSeeds = new int[hashCount];
        
        // 初始化哈希种子
        for (int i = 0; i < hashCount; i++) {
            hashSeeds[i] = i * 2 + 3;
        }
    }
    
    // 添加元素
    public void add(String value) {
        for (int seed : hashSeeds) {
            int index = hash(value, seed) % bitSize;
            bitSet.set(index);
        }
    }
    
    // 判断元素是否存在
    public boolean mightContain(String value) {
        for (int seed : hashSeeds) {
            int index = hash(value, seed) % bitSize;
            if (!bitSet.get(index)) {
                return false;
            }
        }
        return true;
    }
    
    // 哈希函数
    private int hash(String value, int seed) {
        int result = 0;
        for (int i = 0; i < value.length(); i++) {
            result = seed * result + value.charAt(i);
        }
        return Math.abs(result);
    }
}

Redis集成布隆过滤器

@Service
public class CacheService {
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    private BloomFilter bloomFilter = new BloomFilter(1000000, 0.01);
    
    public Object getData(String key) {
        // 1. 布隆过滤器检查
        if (!bloomFilter.mightContain(key)) {
            return null; // 肯定不存在，直接返回
        }
        
        // 2. 缓存查询
        Object cacheData = redisTemplate.opsForValue().get(key);
        if (cacheData != null) {
            return cacheData;
        }
        
        // 3. 数据库查询
        Object dbData = queryFromDatabase(key);
        if (dbData != null) {
            // 4. 缓存数据并更新布隆过滤器
            redisTemplate.opsForValue().set(key, dbData, Duration.ofMinutes(30));
            bloomFilter.add(key);
        } else {
            // 5. 缓存空值防止穿透
            redisTemplate.opsForValue().set(key + ":null", "NULL", Duration.ofMinutes(5));
        }
        
        return dbData;
    }
    
    private Object queryFromDatabase(String key) {
        // 数据库查询逻辑
        return null;
    }
}

2. 缓存空值策略

对于查询结果为空的数据，也缓存一个特殊的标识，避免重复查询数据库。

public class CacheNullValueStrategy {
    private static final String NULL_VALUE = "NULL_CACHE";
    private static final long NULL_CACHE_EXPIRE = 300; // 5分钟
    
    public Object getDataWithNullCache(String key) {
        // 检查是否存在空值缓存
        String nullKey = key + ":null";
        if (redisTemplate.hasKey(nullKey)) {
            return null; // 直接返回空值
        }
        
        // 正常缓存查询
        Object cacheData = redisTemplate.opsForValue().get(key);
        if (cacheData != null) {
            return cacheData;
        }
        
        // 数据库查询
        Object dbData = queryFromDatabase(key);
        if (dbData != null) {
            redisTemplate.opsForValue().set(key, dbData, Duration.ofMinutes(30));
        } else {
            // 缓存空值
            redisTemplate.opsForValue().set(nullKey, NULL_VALUE, Duration.ofSeconds(NULL_CACHE_EXPIRE));
        }
        
        return dbData;
    }
}

缓存击穿解决方案

1. 分布式锁机制

使用分布式锁确保同一时间只有一个线程去查询数据库，其他线程等待结果。

@Service
public class DistributedLockCacheService {
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    public Object getDataWithDistributedLock(String key) {
        // 1. 缓存查询
        Object cacheData = redisTemplate.opsForValue().get(key);
        if (cacheData != null) {
            return cacheData;
        }
        
        // 2. 获取分布式锁
        String lockKey = "lock:" + key;
        String lockValue = UUID.randomUUID().toString();
        Boolean isLocked = redisTemplate.opsForValue().setIfAbsent(
            lockKey, lockValue, Duration.ofSeconds(10)
        );
        
        if (Boolean.TRUE.equals(isLocked)) {
            try {
                // 3. 再次检查缓存（双重检查）
                cacheData = redisTemplate.opsForValue().get(key);
                if (cacheData != null) {
                    return cacheData;
                }
                
                // 4. 查询数据库
                Object dbData = queryFromDatabase(key);
                if (dbData != null) {
                    // 5. 写入缓存
                    redisTemplate.opsForValue().set(key, dbData, Duration.ofMinutes(30));
                }
                return dbData;
            } finally {
                // 6. 释放锁
                releaseLock(lockKey, lockValue);
            }
        } else {
            // 7. 等待并重试
            try {
                Thread.sleep(100);
                return getDataWithDistributedLock(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
        );
    }
    
    private Object queryFromDatabase(String key) {
        // 数据库查询逻辑
        return null;
    }
}

2. 互斥锁优化版本

使用Redis的SET命令原子性操作实现更高效的分布式锁：

public class OptimizedDistributedLock {
    private static final String LOCK_SUCCESS = "OK";
    private static final Long RELEASE_SUCCESS = 1L;
    
    public boolean tryLock(String key, String value, int expireTime) {
        String script = "return redis.call('set', KEYS[1], ARGV[1], 'NX', 'EX', ARGV[2])";
        Object result = redisTemplate.execute(
            new DefaultRedisScript<>(script, String.class),
            Collections.singletonList(key),
            value,
            String.valueOf(expireTime)
        );
        return LOCK_SUCCESS.equals(result);
    }
    
    public boolean releaseLock(String key, String value) {
        String script = "if redis.call('get', KEYS[1]) == ARGV[1] then " +
                       "return redis.call('del', KEYS[1]) else return 0 end";
        Long result = redisTemplate.execute(
            new DefaultRedisScript<>(script, Long.class),
            Collections.singletonList(key),
            value
        );
        return RELEASE_SUCCESS.equals(result);
    }
}

3. 热点数据永不过期策略

对于热点数据，可以采用永不过期的策略，通过后台线程定期更新缓存：

@Component
public class HotDataCacheManager {
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    private final Set<String> hotKeys = new HashSet<>();
    
    // 标记热点key
    public void markAsHotKey(String key) {
        hotKeys.add(key);
        // 设置永不过期
        redisTemplate.persist(key);
    }
    
    // 定时更新热点数据
    @Scheduled(fixedDelay = 300000) // 5分钟执行一次
    public void refreshHotData() {
        for (String key : hotKeys) {
            Object dbData = queryFromDatabase(key);
            if (dbData != null) {
                redisTemplate.opsForValue().set(key, dbData);
            }
        }
    }
    
    private Object queryFromDatabase(String key) {
        // 数据库查询逻辑
        return null;
    }
}

缓存雪崩解决方案

1. 过期时间随机化

为缓存key设置随机的过期时间，避免大量key同时过期：

@Service
public class RandomExpireCacheService {
    private static final int BASE_EXPIRE_TIME = 1800; // 30分钟
    private static final int RANDOM_RANGE = 600;      // ±10分钟
    
    public void setCacheWithRandomExpire(String key, Object value) {
        // 生成随机过期时间
        int randomExpire = BASE_EXPIRE_TIME + 
                          new Random().nextInt(RANDOM_RANGE * 2) - 
                          RANDOM_RANGE;
        
        redisTemplate.opsForValue().set(key, value, Duration.ofSeconds(randomExpire));
    }
    
    // 批量设置缓存
    public void batchSetCacheWithRandomExpire(Map<String, Object> dataMap) {
        dataMap.forEach((key, value) -> {
            setCacheWithRandomExpire(key, value);
        });
    }
}

2. 多级缓存架构

构建多级缓存架构，包括本地缓存、分布式缓存和数据库：

@Component
public class MultiLevelCacheManager {
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    // 本地缓存（一级缓存）
    private final Cache<String, Object> localCache = Caffeine.newBuilder()
        .maximumSize(10000)
        .expireAfterWrite(5, TimeUnit.MINUTES)
        .build();
    
    public Object getData(String key) {
        // 1. 本地缓存查询
        Object localData = localCache.getIfPresent(key);
        if (localData != null) {
            return localData;
        }
        
        // 2. Redis缓存查询
        Object redisData = redisTemplate.opsForValue().get(key);
        if (redisData != null) {
            // 写入本地缓存
            localCache.put(key, redisData);
            return redisData;
        }
        
        // 3. 数据库查询
        Object dbData = queryFromDatabase(key);
        if (dbData != null) {
            // 写入各级缓存
            redisTemplate.opsForValue().set(key, dbData, Duration.ofMinutes(30));
            localCache.put(key, dbData);
        }
        
        return dbData;
    }
    
    // 更新缓存
    public void updateCache(String key, Object value) {
        // 更新数据库
        updateDatabase(key, value);
        
        // 清除各级缓存
        localCache.invalidate(key);
        redisTemplate.delete(key);
    }
    
    private Object queryFromDatabase(String key) {
        // 数据库查询逻辑
        return null;
    }
    
    private void updateDatabase(String key, Object value) {
        // 数据库更新逻辑
    }
}

3. 熔断降级机制

使用Hystrix实现熔断降级，防止缓存失效时系统崩溃：

@Component
public class CircuitBreakerCacheService {
    @HystrixCommand(
        fallbackMethod = "getDataFallback",
        commandProperties = {
            @HystrixProperty(name = "execution.isolation.thread.timeoutInMilliseconds", value = "3000"),
            @HystrixProperty(name = "circuitBreaker.requestVolumeThreshold", value = "10"),
            @HystrixProperty(name = "circuitBreaker.errorThresholdPercentage", value = "50"),
            @HystrixProperty(name = "circuitBreaker.sleepWindowInMilliseconds", value = "5000")
        }
    )
    public Object getData(String key) {
        Object cacheData = redisTemplate.opsForValue().get(key);
        if (cacheData != null) {
            return cacheData;
        }
        
        Object dbData = queryFromDatabase(key);
        if (dbData != null) {
            redisTemplate.opsForValue().set(key, dbData, Duration.ofMinutes(30));
        }
        return dbData;
    }
    
    // 降级方法
    public Object getDataFallback(String key) {
        // 返回默认值或空值
        return getDefaultData(key);
    }
    
    private Object getDefaultData(String key) {
        // 返回默认数据逻辑
        return new HashMap<String, Object>() {{
            put("status", "service_unavailable");
            put("message", "系统繁忙，请稍后重试");
        }};
    }
    
    private Object queryFromDatabase(String key) {
        // 数据库查询逻辑
        return null;
    }
}

高可用缓存架构设计

1. Redis集群架构

构建Redis集群以提高可用性和扩展性：

# redis-cluster.yaml
version: '3.8'
services:
  redis-node-1:
    image: redis:6.2-alpine
    command: redis-server /usr/local/etc/redis/redis.conf
    volumes:
      - ./redis.conf:/usr/local/etc/redis/redis.conf
      - redis-data-1:/data
    ports:
      - "7001:7001"
    networks:
      - redis-cluster

  redis-node-2:
    image: redis:6.2-alpine
    command: redis-server /usr/local/etc/redis/redis.conf
    volumes:
      - ./redis.conf:/usr/local/etc/redis/redis.conf
      - redis-data-2:/data
    ports:
      - "7002:7002"
    networks:
      - redis-cluster

networks:
  redis-cluster:
    driver: bridge

volumes:
  redis-data-1:
  redis-data-2:

2. 缓存监控与告警

实现缓存监控系统，及时发现和处理异常：

@Component
public class CacheMonitor {
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    private final MeterRegistry meterRegistry;
    private final Counter cacheHitCounter;
    private final Counter cacheMissCounter;
    private final Timer cacheOperationTimer;
    
    public CacheMonitor(MeterRegistry meterRegistry) {
        this.meterRegistry = meterRegistry;
        this.cacheHitCounter = Counter.builder("cache.hit")
            .description("缓存命中次数")
            .register(meterRegistry);
        this.cacheMissCounter = Counter.builder("cache.miss")
            .description("缓存未命中次数")
            .register(meterRegistry);
        this.cacheOperationTimer = Timer.builder("cache.operation")
            .description("缓存操作耗时")
            .register(meterRegistry);
    }
    
    public Object getDataWithMonitoring(String key) {
        Timer.Sample sample = Timer.start(meterRegistry);
        
        try {
            Object cacheData = redisTemplate.opsForValue().get(key);
            if (cacheData != null) {
                cacheHitCounter.increment();
                sample.stop(cacheOperationTimer);
                return cacheData;
            } else {
                cacheMissCounter.increment();
                sample.stop(cacheOperationTimer);
                return queryFromDatabase(key);
            }
        } catch (Exception e) {
            sample.stop(cacheOperationTimer);
            throw e;
        }
    }
    
    private Object queryFromDatabase(String key) {
        // 数据库查询逻辑
        return null;
    }
}

3. 缓存预热机制

系统启动时预热热点数据：

@Component
public class CachePreheatService {
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    @EventListener(ApplicationReadyEvent.class)
    public void preheatCache() {
        // 预热配置数据
        preheatConfigData();
        
        // 预热用户数据
        preheatUserData();
        
        // 预热商品数据
        preheatProductData();
    }
    
    private void preheatConfigData() {
        List<Config> configs = configService.getAllConfigs();
        configs.forEach(config -> {
            String key = "config:" + config.getKey();
            redisTemplate.opsForValue().set(key, config.getValue(), 
                Duration.ofHours(24));
        });
    }
    
    private void preheatUserData() {
        List<Long> hotUserIds = userService.getHotUserIds();
        hotUserIds.forEach(userId -> {
            User user = userService.getUserById(userId);
            if (user != null) {
                String key = "user:" + userId;
                redisTemplate.opsForValue().set(key, user, 
                    Duration.ofHours(2));
            }
        });
    }
    
    private void preheatProductData() {
        List<Long> hotProductIds = productService.getHotProductIds();
        hotProductIds.forEach(productId -> {
            Product product = productService.getProductById(productId);
            if (product != null) {
                String key = "product:" + productId;
                redisTemplate.opsForValue().set(key, product, 
                    Duration.ofHours(1));
            }
        });
    }
}

最佳实践总结

1. 缓存设计原则

  1. 缓存粒度适中：不要缓存过大的对象，避免内存浪费
  2. 过期策略合理：根据数据更新频率设置合适的过期时间
  3. 缓存更新及时：确保缓存与数据库数据的一致性
  4. 监控告警完善：建立完善的缓存监控体系

2. 性能优化建议

@Configuration
public class RedisConfig {
    @Bean
    public LettuceConnectionFactory redisConnectionFactory() {
        LettuceClientConfiguration clientConfig = LettuceClientConfiguration.builder()
            .commandTimeout(Duration.ofSeconds(2))
            .shutdownTimeout(Duration.ZERO)
            .build();
        
        RedisStandaloneConfiguration config = new RedisStandaloneConfiguration(
            "localhost", 6379);
        
        return new LettuceConnectionFactory(config, clientConfig);
    }
    
    @Bean
    public RedisTemplate<String, Object> redisTemplate() {
        RedisTemplate<String, Object> template = new RedisTemplate<>();
        template.setConnectionFactory(redisConnectionFactory());
        template.setKeySerializer(new StringRedisSerializer());
        template.setValueSerializer(new GenericJackson2JsonRedisSerializer());
        return template;
    }
}

3. 故障处理机制

@Service
public class FaultTolerantCacheService {
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    // 缓存降级策略
    public Object getDataWithDegradation(String key) {
        try {
            return getData(key);
        } catch (Exception e) {
            // 记录异常日志
            log.error("缓存获取失败，降级处理", e);
            
            // 返回默认值或空值
            return getDefaultValue(key);
        }
    }
    
    // 缓存容错策略
    public Object getDataWithFaultTolerance(String key) {
        try {
            Object cacheData = redisTemplate.opsForValue().get(key);
            if (cacheData != null) {
                return cacheData;
            }
        } catch (Exception e) {
            // 缓存访问失败，直接查询数据库
            log.warn("Redis访问失败，直接查询数据库", e);
        }
        
        return queryFromDatabase(key);
    }
    
    private Object getDefaultValue(String key) {
        // 返回默认值逻辑
        return null;
    }
    
    private Object queryFromDatabase(String key) {
        // 数据库查询逻辑
        return null;
    }
}

总结

Redis缓存系统的三大问题——穿透、击穿、雪崩，需要通过多种技术手段综合解决：

  1. 缓存穿透：通过布隆过滤器、缓存空值等策略，有效拦截无效请求
  2. 缓存击穿：利用分布式锁、互斥锁等机制，确保热点数据的安全访问
  3. 缓存雪崩：采用过期时间随机化、多级缓存、熔断降级等方案，提高系统稳定性

在实际应用中，需要根据业务特点选择合适的解决方案，并建立完善的监控告警机制。通过合理的架构设计和最佳实践，可以构建出高可用、高性能的缓存系统，为业务发展提供强有力的技术支撑。

记住，缓存优化是一个持续的过程，需要不断地监控、分析和调优，才能在复杂多变的生产环境中保持系统的稳定性和高性能。

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

打赏

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

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

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

  1. CVE-2022-0543
  2. docker进阶_dockerswarm
  3. docker进阶_docker-compose
  4. docker基础_网络模式
  5. 命令行清空memcache缓存的两种方法
  6. 百万数据查询优化技巧
  7. laravel 缓存系统的使用
  8. 使用php编程的一些经验总结
  9. BUUCTF-[WebLogic]CVE-2017-10271（反
  10. mysql 配置ssl ，java应用连接
  11. (转载)Go channel 的妙用
  12. MariaDB/MySQL配置文件my.cnf解读
  13. 开发者应该知道的12条Linux系统管理命令
  14. 如何在 Laravel 代码中正确地使用数据库事务
  15. Windows下DOS命令查看占用××端口的程序
  16. Java中的IO操作与流处理

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

发表评论


快捷键：Ctrl+Enter