緩存穿透、擊穿、雪崩：從理論到 Spring Boot 實踐

作者：一安 2025-09-19 07:46:10

存儲存儲架構

在高并發系統中，緩存是提升性能的關鍵組件，但同時也面臨著三大經典難題：緩存穿透、緩存擊穿和緩存雪崩。這些問題如果處理不當，可能導致數據庫壓力驟增，甚至引發系統雪崩。

前言

在高并發系統中，緩存是提升性能的關鍵組件，但同時也面臨著三大經典難題：緩存穿透、緩存擊穿和緩存雪崩。這些問題如果處理不當，可能導致數據庫壓力驟增，甚至引發系統雪崩。

緩存穿透

定義：指查詢一個根本不存在的數據，由于緩存中沒有對應key，所有請求都會穿透到數據庫。

成因：

惡意攻擊：故意請求不存在的key，如用戶ID為負數的查詢
業務邏輯缺陷：誤查不存在的數據
數據已刪除但緩存未清理

危害：

數據庫壓力劇增，可能導致數據庫宕機
系統響應時間變長，影響用戶體驗
浪費服務器資源

緩存擊穿

定義：一個熱點key在緩存中過期的瞬間，有大量并發請求訪問該key，導致所有請求都落到數據庫。

成因：

熱點數據緩存過期
高并發場景下缺乏有效的并發控制

危害：

數據庫瞬間承受巨大壓力
可能導致熱點數據對應的服務不可用
影響關聯業務的正常運行

緩存雪崩

定義：在某一時間段，緩存中大量key集中過期或緩存服務宕機，導致所有請求全部落到數據庫。

成因：

大量key設置了相同的過期時間
緩存服務（如Redis）集群故障
緩存更新機制設計不合理

危害：

數據庫被壓垮，整個系統崩潰
服務可用性急劇下降
可能引發連鎖反應，影響關聯系統

解決方案

Redis 配置類：

@Configuration
public class RedisConfig {

    @Bean
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {
        RedisTemplate<String, Object> template = new RedisTemplate<>();
        template.setConnectionFactory(factory);
        
        // String類型key序列化
        StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
        // 對象序列化
        GenericJackson2JsonRedisSerializer jackson2JsonRedisSerializer = new GenericJackson2JsonRedisSerializer();
        
        template.setKeySerializer(stringRedisSerializer);
        template.setValueSerializer(jackson2JsonRedisSerializer);
        template.setHashKeySerializer(stringRedisSerializer);
        template.setHashValueSerializer(jackson2JsonRedisSerializer);
        
        template.afterPropertiesSet();
        return template;
    }
    
    @Bean
    public RedisCacheManager cacheManager(RedisConnectionFactory factory) {
        // 默認配置
        RedisCacheConfiguration defaultConfig = RedisCacheConfiguration.defaultCacheConfig()
                .entryTtl(Duration.ofMinutes(30)) // 默認過期時間30分鐘
                .serializeKeysWith(RedisSerializationContext.SerializationPair.fromSerializer(new StringRedisSerializer()))
                .serializeValuesWith(RedisSerializationContext.SerializationPair.fromSerializer(new GenericJackson2JsonRedisSerializer()))
                .disableCachingNullValues(); // 默認不緩存null值
        
        // 針對不同緩存名稱設置不同的過期時間
        Map<String, RedisCacheConfiguration> configMap = new HashMap<>();
        configMap.put("userCache", defaultConfig.entryTtl(Duration.ofHours(1)));
        configMap.put("productCache", defaultConfig.entryTtl(Duration.ofMinutes(10)));
        
        return RedisCacheManager.builder(factory)
                .cacheDefaults(defaultConfig)
                .withInitialCacheConfigurations(configMap)
                .build();
    }
}

緩存穿透

方案一：緩存空值

當查詢結果為null時，也將其緩存起來，設置較短的過期時間，防止同一key頻繁穿透到數據庫。

@Service
@RequiredArgsConstructor
public class UserService {

    private final UserMapper userMapper;
    private final RedisTemplate<String, Object> redisTemplate;
    
    // 緩存空值的過期時間設置短一些，如5分鐘
    private static final long NULL_VALUE_TTL = 5;
    
    public User getUserById(Long id) {
        // 1. 先查詢緩存
        String key = "user:" + id;
        User user = (User) redisTemplate.opsForValue().get(key);
        
        // 2. 緩存存在，直接返回
        if (user != null) {
            // 如果是緩存的空對象，返回null
            if (user.getId() == -1L) {
                return null;
            }
            return user;
        }
        
        // 3. 緩存不存在，查詢數據庫
        user = userMapper.selectById(id);
        
        // 4. 數據庫不存在，緩存空對象
        if (user == null) {
            // 使用一個特殊標識表示空值，避免緩存穿透
            user = new User();
            user.setId(-1L); // 特殊標識
            redisTemplate.opsForValue().set(key, user, NULL_VALUE_TTL, TimeUnit.MINUTES);
            return null;
        }
        
        // 5. 數據庫存在，緩存數據
        redisTemplate.opsForValue().set(key, user, 30, TimeUnit.MINUTES);
        return user;
    }
}

方案二：布隆過濾器

布隆過濾器是一種空間效率極高的概率型數據結構，用于判斷一個元素是否在集合中。可以在請求到達緩存層之前，先通過布隆過濾器判斷key是否存在，過濾掉一定不存在的請求。

@Configuration
public class BloomFilterConfig {

    // 預計數據量
    private static final int EXPECTED_INSERTIONS = 1000000;
    // 誤判率，越小需要的空間越大
    private static final double FALSE_POSITIVE_RATE = 0.01;

    @Bean
    public BloomFilter<String> userBloomFilter() {
        // 創建布隆過濾器
        BloomFilter<String> bloomFilter = BloomFilter.create(
                Funnels.stringFunnel(StandardCharsets.UTF_8),
                EXPECTED_INSERTIONS,
                FALSE_POSITIVE_RATE
        );
        
        // 初始化：將已存在的用戶ID添加到布隆過濾器
        // 實際項目中可以從數據庫加載
        // userMapper.findAllIds().forEach(id -> bloomFilter.put("user:" + id));
        
        return bloomFilter;
    }
}

@Service
@RequiredArgsConstructor
public class BloomFilterUserService {

    private final UserMapper userMapper;
    private final RedisTemplate<String, Object> redisTemplate;
    private final BloomFilter<String> userBloomFilter;
    
    public User getUserById(Long id) {
        String key = "user:" + id;
        
        // 1. 先通過布隆過濾器判斷ID是否可能存在
        if (!userBloomFilter.mightContain(key)) {
            // 布隆過濾器判斷不存在，直接返回null
            return null;
        }
        
        // 2. 布隆過濾器判斷可能存在，查詢緩存
        User user = (User) redisTemplate.opsForValue().get(key);
        if (user != null) {
            return user;
        }
        
        // 3. 緩存不存在，查詢數據庫
        user = userMapper.selectById(id);
        if (user == null) {
            return null;
        }
        
        // 4. 緩存數據
        redisTemplate.opsForValue().set(key, user, 30, TimeUnit.MINUTES);
        return user;
    }
}

緩存擊穿

方案一：互斥鎖

當緩存失效時，不是立即去查詢數據庫，而是先嘗試獲取鎖，只有獲取到鎖的線程才去查詢數據庫，其他線程則等待重試。

@Service
@RequiredArgsConstructor
public class ProductService {

    private final ProductMapper productMapper;
    private final RedisTemplate<String, Object> redisTemplate;
    
    // 鎖的過期時間，防止死鎖
    private static final long LOCK_TTL = 5;
    // 緩存過期時間
    private static final long CACHE_TTL = 30;
    
    public Product getProductById(Long id) {
        String key = "product:" + id;
        
        // 1. 先查詢緩存
        Product product = (Product) redisTemplate.opsForValue().get(key);
        if (product != null) {
            return product;
        }
        
        // 2. 緩存不存在，嘗試獲取鎖
        String lockKey = "lock:product:" + id;
        try {
            // 嘗試獲取鎖，setIfAbsent等價于Redis的SETNX命令
            Boolean locked = redisTemplate.opsForValue().setIfAbsent(lockKey, "1", LOCK_TTL, TimeUnit.SECONDS);
            
            if (Boolean.TRUE.equals(locked)) {
                // 3. 獲取到鎖，查詢數據庫
                product = productMapper.selectById(id);
                if (product != null) {
                    // 4. 緩存數據
                    redisTemplate.opsForValue().set(key, product, CACHE_TTL, TimeUnit.MINUTES);
                }
                return product;
            } else {
                // 5. 未獲取到鎖，等待一段時間后重試
                TimeUnit.MILLISECONDS.sleep(50);
                return getProductById(id); // 遞歸重試
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return null;
        } finally {
            // 6. 釋放鎖
            if (Boolean.TRUE.equals(redisTemplate.hasKey(lockKey))) {
                redisTemplate.delete(lockKey);
            }
        }
    }
}

方案二：熱點數據永不過期

對于特別熱點的數據，可以設置為永不過期，通過后臺線程定期更新緩存數據，避免緩存過期導致的擊穿問題。

@Service
@RequiredArgsConstructor
public class HotProductService {

    private final ProductMapper productMapper;
    private final RedisTemplate<String, Object> redisTemplate;
    
    // 熱點商品ID列表
    private static final List<Long> HOT_PRODUCT_IDS = List.of(1001L, 1002L, 1003L);
    
    // 1. 查詢熱點商品，緩存永不過期
    public Product getHotProductById(Long id) {
        // 檢查是否為熱點商品
        if (!HOT_PRODUCT_IDS.contains(id)) {
            throw new IllegalArgumentException("不是熱點商品");
        }
        
        String key = "hot_product:" + id;
        
        // 先查詢緩存
        Product product = (Product) redisTemplate.opsForValue().get(key);
        if (product != null) {
            return product;
        }
        
        // 緩存不存在，查詢數據庫并緩存（永不過期）
        product = productMapper.selectById(id);
        if (product != null) {
            // 設置為永不過期（實際可以設置一個很大的過期時間）
            redisTemplate.opsForValue().set(key, product);
        }
        return product;
    }
    
    // 2. 定時任務更新熱點商品緩存，每10分鐘執行一次
    @Scheduled(fixedRate = 10 * 60 * 1000)
    public void refreshHotProductCache() {
        for (Long productId : HOT_PRODUCT_IDS) {
            Product product = productMapper.selectById(productId);
            if (product != null) {
                String key = "hot_product:" + productId;
                redisTemplate.opsForValue().set(key, product);
            }
        }
    }
}

緩存雪崩

方案一：過期時間隨機化

為不同的key設置隨機的過期時間，避免大量key在同一時間點過期。

@Service
@RequiredArgsConstructor
public class OrderService {

    private final OrderMapper orderMapper;
    private final RedisTemplate<String, Object> redisTemplate;
    
    // 基礎過期時間：30分鐘
    private static final long BASE_TTL = 30;
    // 隨機過期時間范圍：0-10分鐘
    private static final int RANDOM_TTL_RANGE = 10;
    
    private final Random random = new Random();
    
    public Order getOrderById(Long id) {
        String key = "order:" + id;
        
        // 1. 查詢緩存
        Order order = (Order) redisTemplate.opsForValue().get(key);
        if (order != null) {
            return order;
        }
        
        // 2. 緩存不存在，查詢數據庫
        order = orderMapper.selectById(id);
        if (order == null) {
            return null;
        }
        
        // 3. 計算隨機過期時間，避免大量key同時過期
        long randomTTL = random.nextInt(RANDOM_TTL_RANGE);
        long ttl = BASE_TTL + randomTTL;
        
        // 4. 緩存數據
        redisTemplate.opsForValue().set(key, order, ttl, TimeUnit.MINUTES);
        return order;
    }
}

方案二：多級緩存

使用本地緩存（如Caffeine）+ 分布式緩存（如Redis）的多級緩存架構，即使分布式緩存失效，本地緩存也能提供一定的緩沖。

@Configuration
public class MultiLevelCacheConfig {

    // 1. 本地緩存配置（Caffeine）
    @Bean
    public CacheManager caffeineCacheManager() {
        CaffeineCacheManager cacheManager = new CaffeineCacheManager();
        // 設置緩存過期時間：5分鐘
        cacheManager.setCaffeine(Caffeine.newBuilder()
                .expireAfterWrite(5, TimeUnit.MINUTES)
                .maximumSize(10000)); // 最大緩存數量
        return cacheManager;
    }
    
    // 2. 分布式緩存配置（Redis）已在前面的RedisConfig中定義
    
    // 3. 自定義多級緩存管理器（組合本地緩存和Redis緩存）
    @Bean
    @Primary
    public CacheManager multiLevelCacheManager(CacheManager caffeineCacheManager, CacheManager redisCacheManager) {
        return new MultiLevelCacheManager(caffeineCacheManager, redisCacheManager);
    }
}

public class MultiLevelCacheManager implements CacheManager {

    private final CacheManager localCacheManager;
    private final CacheManager redisCacheManager;
    
    public MultiLevelCacheManager(CacheManager localCacheManager, CacheManager redisCacheManager) {
        this.localCacheManager = localCacheManager;
        this.redisCacheManager = redisCacheManager;
    }
    
    @Override
    @NonNull
    public Cache getCache(String name) {
        // 返回組合了本地緩存和Redis緩存的Cache實現
        return new MultiLevelCache(
                localCacheManager.getCache(name),
                redisCacheManager.getCache(name)
        );
    }
    
    @Override
    @NonNull
    public Iterable<String> getCacheNames() {
        return redisCacheManager.getCacheNames();
    }
}

public class MultiLevelCache implements Cache {

    private final Cache localCache;
    private final Cache redisCache;
    
    public MultiLevelCache(Cache localCache, Cache redisCache) {
        this.localCache = localCache;
        this.redisCache = redisCache;
    }
    
    @Override
    public String getName() {
        return redisCache.getName();
    }
    
    @Override
    public Object getNativeCache() {
        return this;
    }
    
    @Override
    @Nullable
    public ValueWrapper get(Object key) {
        // 1. 先查詢本地緩存
        ValueWrapper localValue = localCache.get(key);
        if (localValue != null) {
            return localValue;
        }
        
        // 2. 本地緩存沒有，查詢Redis緩存
        ValueWrapper redisValue = redisCache.get(key);
        if (redisValue != null) {
            // 3. 將Redis緩存的值同步到本地緩存
            localCache.put(key, redisValue.get());
        }
        
        return redisValue;
    }
    
    @Override
    @Nullable
    @SuppressWarnings("unchecked")
    public <T> T get(Object key, @Nullable Class<T> type) {
        // 1. 先查詢本地緩存
        T localValue = localCache.get(key, type);
        if (localValue != null) {
            return localValue;
        }
        
        // 2. 本地緩存沒有，查詢Redis緩存
        T redisValue = redisCache.get(key, type);
        if (redisValue != null) {
            // 3. 將Redis緩存的值同步到本地緩存
            localCache.put(key, redisValue);
        }
        
        return redisValue;
    }
    
    @Override
    @Nullable
    public <T> T get(Object key, Callable<T> valueLoader) {
        // 1. 先查詢本地緩存
        try {
            T localValue = localCache.get(key, valueLoader);
            return localValue;
        } catch (Exception e) {
            // 本地緩存沒有，繼續查詢Redis
        }
        
        // 2. 查詢Redis緩存
        T redisValue = redisCache.get(key, valueLoader);
        // 3. 同步到本地緩存
        localCache.put(key, redisValue);
        
        return redisValue;
    }
    
    @Override
    public void put(Object key, @Nullable Object value) {
        // 同時更新本地緩存和Redis緩存
        localCache.put(key, value);
        redisCache.put(key, value);
    }
    
    @Override
    @Nullable
    public ValueWrapper putIfAbsent(Object key, @Nullable Object value) {
        // 同時更新本地緩存和Redis緩存
        localCache.putIfAbsent(key, value);
        return redisCache.putIfAbsent(key, value);
    }
    
    @Override
    public void evict(Object key) {
        // 同時刪除本地緩存和Redis緩存
        localCache.evict(key);
        redisCache.evict(key);
    }
    
    @Override
    public void clear() {
        // 同時清空本地緩存和Redis緩存
        localCache.clear();
        redisCache.clear();
    }
}

方案三：緩存降級與熔斷

當緩存服務出現異常時，通過降級策略返回默認數據或提示信息，避免請求全部落到數據庫。可以使用Sentinel或Hystrix實現熔斷降級。

@Service
@RequiredArgsConstructor
public class DegradedProductService {

    private final ProductMapper productMapper;
    private final RedisTemplate<String, Object> redisTemplate;
    
    // 使用Sentinel注解實現熔斷降級
    @SentinelResource(
        value = "getProductById",
        blockHandler = "handleGetProductBlocked", // 限流/熔斷時的處理方法
        fallback = "handleGetProductFallback" // 拋出異常時的處理方法
    )
    public Product getProductById(Long id) {
        String key = "product:" + id;
        
        // 1. 查詢緩存
        Product product = (Product) redisTemplate.opsForValue().get(key);
        if (product != null) {
            return product;
        }
        
        // 2. 緩存不存在，查詢數據庫
        product = productMapper.selectById(id);
        if (product != null) {
            // 3. 緩存數據，設置隨機過期時間
            long ttl = 30 + (long) (Math.random() * 10);
            redisTemplate.opsForValue().set(key, product, ttl, TimeUnit.MINUTES);
        }
        return product;
    }
    
    // 限流/熔斷時的降級處理
    public Product handleGetProductBlocked(Long id, BlockException e) {
        // 可以返回緩存的舊數據、默認數據或提示信息
        Product defaultProduct = new Product();
        defaultProduct.setId(id);
        defaultProduct.setName("服務繁忙，請稍后再試");
        return defaultProduct;
    }
    
    // 異常時的降級處理
    public Product handleGetProductFallback(Long id, Throwable e) {
        // 可以嘗試從本地緩存獲取，或返回兜底數據
        return getLocalCacheProduct(id);
    }
    
    // 本地緩存作為最后的兜底
    private Product getLocalCacheProduct(Long id) {
        // 實際項目中可以使用Caffeine等本地緩存
        return null;
    }
}

總結

最佳實踐

分層防御：結合多種方案解決同一問題，如同時使用布隆過濾器和緩存空值防止穿透
監控告警：實時監控緩存命中率、數據庫壓力等指標，及時發現問題
限流保護：對接口進行限流，防止惡意攻擊和流量突增
灰度發布：緩存策略變更時采用灰度發布，避免大規模影響
災備演練：定期進行緩存失效演練，檢驗系統的容錯能力

責任編輯：武曉燕來源：一安未來