May 12 2021

Redis缓存穿透、击穿、雪崩解决方案：分布式缓存高可用架构设计与实现

作者： 绝缘体..   发布： 2021-05-12 11:21   分类： 未分类   阅读：   抢沙发  
 
更多

Redis缓存穿透、击穿、雪崩解决方案：分布式缓存高可用架构设计与实现

引言

在现代分布式系统中，Redis作为主流的缓存解决方案，承担着提升系统性能、减轻数据库压力的重要职责。然而，在实际应用过程中，缓存相关的三大经典问题——缓存穿透、缓存击穿、缓存雪崩——往往会严重影响系统的稳定性和用户体验。本文将深入分析这些问题的本质，并提供切实可行的解决方案，帮助构建高可用的分布式缓存架构。

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

什么是缓存穿透

缓存穿透是指查询一个根本不存在的数据，由于缓存层和存储层都没有命中，每次请求都会直接打到数据库上。这种情况通常发生在恶意攻击或数据热点不均的情况下，会导致数据库压力骤增，严重时可能造成数据库宕机。

缓存穿透的危害

// 模拟缓存穿透场景
public class CachePenetrationDemo {
    private static final String CACHE_KEY = "user_info:";
    
    public User getUserById(Long userId) {
        // 从缓存获取用户信息
        String cacheValue = redisTemplate.opsForValue().get(CACHE_KEY + userId);
        
        if (cacheValue != null) {
            return JSON.parseObject(cacheValue, User.class);
        }
        
        // 缓存未命中，查询数据库
        User user = userDao.findById(userId);
        
        if (user == null) {
            // 数据库也未找到，缓存空值
            redisTemplate.opsForValue().set(CACHE_KEY + userId, "", 300, TimeUnit.SECONDS);
            return null;
        }
        
        // 缓存用户信息
        redisTemplate.opsForValue().set(CACHE_KEY + userId, JSON.toJSONString(user), 3600, TimeUnit.SECONDS);
        return user;
    }
}

如上代码所示，当查询一个不存在的用户ID时，会频繁访问数据库，造成资源浪费。

解决方案一：布隆过滤器

布隆过滤器是一种概率型数据结构，可以快速判断一个元素是否存在于集合中。通过在缓存层之前加入布隆过滤器，可以有效拦截不存在的请求。

@Component
public class BloomFilterCache {
    private static final String BLOOM_FILTER_KEY = "bloom_filter";
    private static final int CAPACITY = 1000000;
    private static final double ERROR_RATE = 0.01;
    
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    private BloomFilter<String> bloomFilter;
    
    @PostConstruct
    public void init() {
        // 初始化布隆过滤器
        this.bloomFilter = BloomFilter.create(
            Funnels.stringFunnel(Charsets.UTF_8),
            CAPACITY,
            ERROR_RATE
        );
        
        // 加载已存在的数据到布隆过滤器
        loadExistingData();
    }
    
    public boolean existsInBloomFilter(String key) {
        return bloomFilter.mightContain(key);
    }
    
    public void addKeyToBloomFilter(String key) {
        bloomFilter.put(key);
    }
    
    public User getUserWithBloomFilter(Long userId) {
        String key = "user:" + userId;
        
        // 先检查布隆过滤器
        if (!existsInBloomFilter(key)) {
            return null; // 布隆过滤器判断不存在，直接返回
        }
        
        // 布隆过滤器可能存在，继续查询缓存
        String cacheValue = redisTemplate.opsForValue().get(key);
        if (cacheValue != null) {
            return JSON.parseObject(cacheValue, User.class);
        }
        
        // 缓存未命中，查询数据库
        User user = userDao.findById(userId);
        if (user != null) {
            // 存在数据，加入布隆过滤器
            addKeyToBloomFilter(key);
            redisTemplate.opsForValue().set(key, JSON.toJSONString(user), 3600, TimeUnit.SECONDS);
        } else {
            // 不存在数据，缓存空值
            redisTemplate.opsForValue().set(key, "", 300, TimeUnit.SECONDS);
        }
        
        return user;
    }
}

解决方案二：缓存空值

对于查询结果为空的情况，同样需要进行缓存处理，避免重复查询数据库。

@Service
public class UserService {
    private static final String CACHE_KEY_PREFIX = "user_info:";
    private static final long CACHE_NULL_TTL = 300; // 5分钟
    private static final long CACHE_NORMAL_TTL = 3600; // 1小时
    
    public User getUserById(Long userId) {
        String cacheKey = CACHE_KEY_PREFIX + userId;
        
        // 先从缓存获取
        String cacheValue = redisTemplate.opsForValue().get(cacheKey);
        
        if (cacheValue != null) {
            if ("".equals(cacheValue)) {
                // 空值缓存，直接返回null
                return null;
            }
            return JSON.parseObject(cacheValue, User.class);
        }
        
        // 缓存未命中，查询数据库
        User user = userDao.findById(userId);
        
        if (user == null) {
            // 数据库未找到，缓存空值
            redisTemplate.opsForValue().set(cacheKey, "", CACHE_NULL_TTL, TimeUnit.SECONDS);
        } else {
            // 数据库找到，缓存正常数据
            redisTemplate.opsForValue().set(cacheKey, JSON.toJSONString(user), CACHE_NORMAL_TTL, TimeUnit.SECONDS);
        }
        
        return user;
    }
}

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

什么是缓存击穿

缓存击穿是指某个热点key在缓存过期后，大量并发请求同时访问该key，导致数据库瞬间压力剧增的现象。与缓存穿透不同的是，这个key在数据库中是真实存在的。

缓存击穿的典型场景

// 缓存击穿示例
@RestController
public class ProductController {
    private static final String PRODUCT_CACHE_KEY = "product:";
    
    @GetMapping("/product/{id}")
    public ResponseEntity<Product> getProduct(@PathVariable Long id) {
        String cacheKey = PRODUCT_CACHE_KEY + id;
        
        // 从缓存获取商品信息
        String cachedProduct = redisTemplate.opsForValue().get(cacheKey);
        
        if (cachedProduct != null) {
            return ResponseEntity.ok(JSON.parseObject(cachedProduct, Product.class));
        }
        
        // 缓存失效，直接查询数据库
        Product product = productService.getProductById(id);
        
        if (product != null) {
            // 重新缓存
            redisTemplate.opsForValue().set(cacheKey, JSON.toJSONString(product), 3600, TimeUnit.SECONDS);
        }
        
        return ResponseEntity.ok(product);
    }
}

解决方案一：互斥锁机制

通过分布式锁确保同一时间只有一个线程去查询数据库并更新缓存。

@Service
public class ProductService {
    private static final String PRODUCT_CACHE_KEY = "product:";
    private static final String LOCK_KEY_PREFIX = "lock:product:";
    private static final long CACHE_TTL = 3600;
    private static final long LOCK_EXPIRE_TIME = 10; // 锁超时时间
    
    public Product getProductById(Long productId) {
        String cacheKey = PRODUCT_CACHE_KEY + productId;
        String lockKey = LOCK_KEY_PREFIX + productId;
        
        // 先尝试从缓存获取
        String cachedProduct = redisTemplate.opsForValue().get(cacheKey);
        if (cachedProduct != null) {
            return JSON.parseObject(cachedProduct, Product.class);
        }
        
        // 尝试获取分布式锁
        String lockValue = UUID.randomUUID().toString();
        Boolean lockAcquired = redisTemplate.opsForValue()
            .setIfAbsent(lockKey, lockValue, LOCK_EXPIRE_TIME, TimeUnit.SECONDS);
        
        try {
            if (lockAcquired) {
                // 获取锁成功，再次检查缓存
                cachedProduct = redisTemplate.opsForValue().get(cacheKey);
                if (cachedProduct != null) {
                    return JSON.parseObject(cachedProduct, Product.class);
                }
                
                // 缓存未命中，查询数据库
                Product product = productDao.findById(productId);
                
                if (product != null) {
                    // 更新缓存
                    redisTemplate.opsForValue().set(cacheKey, JSON.toJSONString(product), CACHE_TTL, TimeUnit.SECONDS);
                } else {
                    // 数据库无数据，设置空值缓存
                    redisTemplate.opsForValue().set(cacheKey, "", 300, TimeUnit.SECONDS);
                }
                
                return product;
            } else {
                // 获取锁失败，等待一段时间后重试
                Thread.sleep(100);
                return getProductById(productId); // 递归重试
            }
        } catch (Exception e) {
            throw new RuntimeException("获取商品信息失败", e);
        } finally {
            // 释放锁
            releaseLock(lockKey, lockValue);
        }
    }
    
    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);
    }
}

解决方案二：随机过期时间

为热点数据设置随机的过期时间，避免集中过期。

@Service
public class HotKeyService {
    private static final String HOT_KEY_PREFIX = "hot_key:";
    private static final long BASE_TTL = 3600;
    private static final int TTL_RANGE = 300; // 随机范围5分钟
    
    public String getHotKeyData(String key) {
        String cacheKey = HOT_KEY_PREFIX + key;
        
        // 先从缓存获取
        String value = redisTemplate.opsForValue().get(cacheKey);
        if (value != null) {
            return value;
        }
        
        // 缓存未命中，查询数据库
        String dbValue = databaseService.getData(key);
        
        if (dbValue != null) {
            // 设置随机过期时间，避免集中过期
            long randomTtl = BASE_TTL + new Random().nextInt(TTL_RANGE);
            redisTemplate.opsForValue().set(cacheKey, dbValue, randomTtl, TimeUnit.SECONDS);
        }
        
        return dbValue;
    }
}

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

什么是缓存雪崩

缓存雪崩是指在某一时刻大量缓存同时失效，导致所有请求都直接访问数据库，造成数据库压力骤增甚至宕机的现象。这通常是由于缓存设置相同的过期时间造成的。

缓存雪崩的影响

// 缓存雪崩示例
@Component
public class CacheAvalancheDemo {
    private static final String CACHE_KEY = "data:";
    private static final long DEFAULT_TTL = 3600; // 1小时
    
    public String getData(String key) {
        String cacheKey = CACHE_KEY + key;
        
        // 所有数据在同一时间过期
        String value = redisTemplate.opsForValue().get(cacheKey);
        if (value != null) {
            return value;
        }
        
        // 大量请求同时访问数据库
        String dbValue = databaseService.getValue(key);
        redisTemplate.opsForValue().set(cacheKey, dbValue, DEFAULT_TTL, TimeUnit.SECONDS);
        
        return dbValue;
    }
}

解决方案一：多级缓存架构

构建多层次的缓存体系，降低单一缓存层的压力。

@Component
public class MultiLevelCache {
    private static final String LOCAL_CACHE_KEY = "local_cache:";
    private static final String REDIS_CACHE_KEY = "redis_cache:";
    private static final String DB_KEY = "database:";
    
    // 本地缓存
    private final LoadingCache<String, String> localCache = 
        Caffeine.newBuilder()
            .maximumSize(1000)
            .expireAfterWrite(300, TimeUnit.SECONDS)
            .build(key -> fetchFromRedis(key));
    
    // Redis缓存
    @Autowired
    private RedisTemplate<String, String> redisTemplate;
    
    // 数据库访问
    @Autowired
    private DatabaseService databaseService;
    
    public String getData(String key) {
        // 1. 先查本地缓存
        String localValue = localCache.getIfPresent(key);
        if (localValue != null) {
            return localValue;
        }
        
        // 2. 查Redis缓存
        String redisValue = redisTemplate.opsForValue().get(REDIS_CACHE_KEY + key);
        if (redisValue != null) {
            // 同步到本地缓存
            localCache.put(key, redisValue);
            return redisValue;
        }
        
        // 3. 查询数据库
        String dbValue = databaseService.getValue(key);
        if (dbValue != null) {
            // 写入多级缓存
            redisTemplate.opsForValue().set(REDIS_CACHE_KEY + key, dbValue, 3600, TimeUnit.SECONDS);
            localCache.put(key, dbValue);
        }
        
        return dbValue;
    }
    
    // 定期刷新缓存
    @Scheduled(fixedRate = 60000) // 每分钟执行一次
    public void refreshCache() {
        // 可以在这里实现缓存预热逻辑
    }
}

解决方案二：缓存预热机制

在业务高峰期前预先加载热点数据到缓存中。

@Component
public class CacheWarmupService {
    private static final String WARMUP_KEY_PREFIX = "warmup:";
    private static final long WARMUP_TTL = 7200; // 2小时
    
    @Autowired
    private RedisTemplate<String, String> redisTemplate;
    
    @Autowired
    private ProductService productService;
    
    @PostConstruct
    public void warmUpCache() {
        // 系统启动时预热热点数据
        List<Long> hotProductIds = getHotProductIds();
        
        for (Long productId : hotProductIds) {
            try {
                Product product = productService.getProductById(productId);
                if (product != null) {
                    String cacheKey = WARMUP_KEY_PREFIX + "product:" + productId;
                    redisTemplate.opsForValue().set(cacheKey, JSON.toJSONString(product), WARMUP_TTL, TimeUnit.SECONDS);
                }
            } catch (Exception e) {
                log.error("缓存预热失败，productId: {}", productId, e);
            }
        }
    }
    
    @Scheduled(cron = "0 0 2 * * ?") // 每天凌晨2点执行
    public void scheduledWarmUp() {
        // 定期预热
        warmUpCache();
    }
    
    private List<Long> getHotProductIds() {
        // 获取热门商品ID列表
        return Arrays.asList(1L, 2L, 3L, 4L, 5L);
    }
}

高可用缓存架构设计

架构图设计

# 缓存架构配置示例
cache:
  cluster:
    nodes:
      - host: 192.168.1.10
        port: 6379
        role: master
      - host: 192.168.1.11
        port: 6379
        role: slave
      - host: 192.168.1.12
        port: 6379
        role: slave
    sentinel:
      hosts:
        - host: 192.168.1.20
          port: 26379
        - host: 192.168.1.21
          port: 26379
        - host: 192.168.1.22
          port: 26379
  config:
    max-active: 200
    max-idle: 50
    min-idle: 10
    timeout: 2000
    health-check: true

缓存监控与告警

@Component
public class CacheMonitor {
    private static final Logger logger = LoggerFactory.getLogger(CacheMonitor.class);
    
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    @Scheduled(fixedRate = 30000) // 每30秒检查一次
    public void monitorCacheHealth() {
        try {
            // 检查连接状态
            String pingResult = redisTemplate.getConnectionFactory()
                .getConnection().ping();
            
            if (!"PONG".equals(pingResult)) {
                logger.warn("Redis连接异常");
                // 发送告警
                sendAlert("Redis连接异常");
            }
            
            // 检查内存使用率
            String info = redisTemplate.getConnectionFactory()
                .getConnection().info("memory");
            
            // 分析内存使用情况
            parseMemoryInfo(info);
            
        } catch (Exception e) {
            logger.error("缓存监控异常", e);
        }
    }
    
    private void parseMemoryInfo(String info) {
        // 解析Redis内存信息
        String[] lines = info.split("\n");
        for (String line : lines) {
            if (line.startsWith("used_memory:")) {
                long usedMemory = Long.parseLong(line.split(":")[1]);
                if (usedMemory > 800000000) { // 800MB
                    logger.warn("Redis内存使用率过高: {} bytes", usedMemory);
                    sendAlert("Redis内存使用率过高");
                }
            }
        }
    }
    
    private void sendAlert(String message) {
        // 实现告警通知逻辑
        // 可以集成邮件、短信、钉钉等告警方式
        System.out.println("告警: " + message);
    }
}

生产环境部署最佳实践

Redis集群配置

# application-prod.properties
spring.redis.cluster.nodes=192.168.1.10:6379,192.168.1.11:6379,192.168.1.12:6379
spring.redis.cluster.max-redirects=3
spring.redis.cluster.timeout=2000ms
spring.redis.cluster.max-active=200
spring.redis.cluster.max-idle=50
spring.redis.cluster.min-idle=10
spring.redis.cluster.test-on-borrow=true
spring.redis.cluster.test-on-return=true

缓存策略配置

@Configuration
@EnableConfigurationProperties(CacheProperties.class)
public class CacheConfig {
    
    @Bean
    public CacheManager cacheManager(RedisConnectionFactory connectionFactory) {
        RedisCacheConfiguration config = RedisCacheConfiguration.defaultCacheConfig()
            .entryTtl(Duration.ofHours(1))
            .serializeKeysWith(RedisSerializationContext.SerializationPair.fromSerializer(
                new StringRedisSerializer()))
            .serializeValuesWith(RedisSerializationContext.SerializationPair.fromSerializer(
                new GenericJackson2JsonRedisSerializer()))
            .disableCachingNullValues();
        
        return RedisCacheManager.builder(connectionFactory)
            .withInitialCacheConfigurations(Collections.singletonMap(
                "default", config))
            .build();
    }
    
    @Bean
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory connectionFactory) {
        RedisTemplate<String, Object> template = new RedisTemplate<>();
        template.setConnectionFactory(connectionFactory);
        template.setKeySerializer(new StringRedisSerializer());
        template.setValueSerializer(new GenericJackson2JsonRedisSerializer());
        template.afterPropertiesSet();
        return template;
    }
}

性能调优建议

# Redis性能调优参数
# redis.conf
tcp-keepalive 300
timeout 0
tcp-nodelay yes
maxmemory 2gb
maxmemory-policy allkeys-lru
hash-max-ziplist-entries 512
hash-max-ziplist-value 64
list-max-ziplist-size -2
list-compress-depth 0
zset-max-ziplist-entries 128
zset-max-ziplist-value 64

总结与展望

通过本文的详细分析和实践方案，我们可以看到缓存穿透、击穿、雪崩问题是分布式系统中必须重视的核心挑战。有效的解决方案包括：

  1. 缓存穿透：采用布隆过滤器和空值缓存机制
  2. 缓存击穿：使用互斥锁和随机过期时间
  3. 缓存雪崩：构建多级缓存架构和实施缓存预热

这些技术手段的综合运用能够显著提升系统的稳定性和性能表现。在实际生产环境中，还需要结合具体的业务场景和系统特点，持续优化缓存策略，建立完善的监控告警体系，确保分布式缓存架构的高可用性。

未来随着微服务架构的进一步发展和云原生技术的普及，缓存技术也将朝着更加智能化、自动化的方向演进。我们需要持续关注新技术的发展，不断优化和完善我们的缓存架构设计。

本文来自极简博客，作者：热血少年，转载请注明原文链接：Redis缓存穿透、击穿、雪崩解决方案：分布式缓存高可用架构设计与实现

打赏

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

该日志由 绝缘体.. 于 2021年05月12日 发表在 未分类 分类下， 你可以发表评论，并在保留原文地址及作者的情况下引用到你的网站或博客。
原创文章转载请注明: 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