发布于 由admin

Spring Boot并发请求导致缓存雪崩的解决机制与架构方案

Spring Boot并发请求导致缓存雪崩的解决机制与架构方案

大家好，今天我们来聊聊Spring Boot应用中一个常见且棘手的问题：并发请求导致缓存雪崩。缓存雪崩指的是在短时间内，缓存中大量的key同时过期或失效，导致大量的请求涌向数据库，最终压垮数据库的现象。这对于高并发系统来说，是灾难性的。

我们将会深入探讨缓存雪崩的成因、危害，以及如何在Spring Boot应用中构建完善的架构方案来预防和应对缓存雪崩。

1. 缓存雪崩的成因与危害

缓存雪崩通常由以下几个原因导致：

  • 大量Key同时过期: 这是最常见的原因。如果大量的key设置了相同的过期时间，那么在过期时刻，这些key会同时失效，导致大量请求直接落到数据库上。
  • 缓存服务宕机: 如果缓存服务（如Redis）突然宕机，所有缓存失效，请求全部涌向数据库，造成雪崩。
  • 热点Key失效: 当一个被频繁访问的热点key过期后，大量的并发请求会同时请求数据库来更新该key，造成数据库压力过大。

缓存雪崩的危害是显而易见的：

  • 数据库压力剧增: 数据库可能会因为过载而崩溃，导致服务不可用。
  • 服务响应时间延长: 数据库响应变慢，直接导致应用响应时间延长，用户体验下降。
  • 系统可用性降低: 在极端情况下，整个系统可能会崩溃，导致服务完全不可用。

2. Spring Boot中预防缓存雪崩的策略与实现

接下来，我们将介绍Spring Boot中预防缓存雪崩的几种常用策略，并给出相应的代码示例。

2.1 避免Key同时过期：过期时间分散化

最简单有效的策略是避免大量的key同时过期。我们可以为不同的key设置不同的过期时间，让它们分散在不同的时间段过期。

import org.springframework.stereotype.Component;
import java.util.Random;

@Component
public class CacheKeyGenerator {

    private static final int DEFAULT_EXPIRATION_SECONDS = 3600; // 默认过期时间1小时
    private static final int EXPIRATION_VARIANCE = 300; // 随机波动范围5分钟

    public int generateExpiration() {
        Random random = new Random();
        // 在默认过期时间的基础上，随机增加或减少一段时间
        return DEFAULT_EXPIRATION_SECONDS + random.nextInt(EXPIRATION_VARIANCE) - EXPIRATION_VARIANCE / 2;
    }

    public int generateExpiration(int baseExpiration) {
        Random random = new Random();
        // 在baseExpiration基础上，随机增加或减少一段时间
        return baseExpiration + random.nextInt(EXPIRATION_VARIANCE) - EXPIRATION_VARIANCE / 2;
    }
}
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;

import java.util.concurrent.TimeUnit;

@Service
public class ProductService {

    @Autowired
    private StringRedisTemplate redisTemplate;

    @Autowired
    private CacheKeyGenerator cacheKeyGenerator;

    public String getProductInfo(String productId) {
        String key = "product:" + productId;
        String productInfo = redisTemplate.opsForValue().get(key);

        if (productInfo == null) {
            // 从数据库获取数据
            productInfo = loadProductFromDatabase(productId);

            if (productInfo != null) {
                // 设置缓存，过期时间随机化
                int expiration = cacheKeyGenerator.generateExpiration();
                redisTemplate.opsForValue().set(key, productInfo, expiration, TimeUnit.SECONDS);
            }
        }

        return productInfo;
    }

    private String loadProductFromDatabase(String productId) {
        // 模拟从数据库加载数据
        try {
            Thread.sleep(100); // 模拟数据库查询耗时
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "Product Info for " + productId;
    }
}

说明: CacheKeyGenerator类负责生成随机的过期时间，在默认过期时间的基础上增加一个随机的波动范围。ProductService类在将数据写入Redis时，使用CacheKeyGenerator生成的过期时间。

2.2 使用互斥锁(Mutex Lock)防止缓存击穿

当一个热点key失效时，大量的并发请求会同时查询数据库来更新该key，造成缓存击穿。可以使用互斥锁来解决这个问题，保证只有一个线程能够查询数据库并更新缓存，其他线程等待。

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;

import java.util.concurrent.TimeUnit;

@Service
public class ProductService {

    @Autowired
    private StringRedisTemplate redisTemplate;

    private static final String LOCK_PREFIX = "lock:product:";

    public String getProductInfoWithLock(String productId) {
        String key = "product:" + productId;
        String productInfo = redisTemplate.opsForValue().get(key);

        if (productInfo == null) {
            // 尝试获取锁
            String lockKey = LOCK_PREFIX + productId;
            Boolean locked = redisTemplate.opsForValue().setIfAbsent(lockKey, "locked", 10, TimeUnit.SECONDS); // 设置10秒过期时间，防止死锁

            if (locked != null && locked) {
                try {
                    // 获取到锁，查询数据库并更新缓存
                    productInfo = loadProductFromDatabase(productId);

                    if (productInfo != null) {
                        redisTemplate.opsForValue().set(key, productInfo, 3600, TimeUnit.SECONDS); // 缓存1小时
                    }
                } finally {
                    // 释放锁
                    redisTemplate.delete(lockKey);
                }
            } else {
                // 没有获取到锁，等待一段时间后重试
                try {
                    Thread.sleep(50);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                return getProductInfoWithLock(productId); // 递归调用，重试
            }
        }

        return productInfo;
    }

    private String loadProductFromDatabase(String productId) {
        // 模拟从数据库加载数据
        try {
            Thread.sleep(100); // 模拟数据库查询耗时
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "Product Info for " + productId;
    }
}

说明: getProductInfoWithLock方法首先尝试从缓存中获取数据，如果缓存不存在，则尝试获取一个分布式锁。只有获取到锁的线程才能查询数据库并更新缓存，其他线程则等待一段时间后重试。

2.3 使用双重检测锁 (Double-Check Locking)

双重检测锁是互斥锁的一种优化，它在加锁之前先进行一次缓存检查，减少不必要的加锁操作。

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;

import java.util.concurrent.TimeUnit;

@Service
public class ProductService {

    @Autowired
    private StringRedisTemplate redisTemplate;

    private static final String LOCK_PREFIX = "lock:product:";

    public String getProductInfoWithDoubleCheck(String productId) {
        String key = "product:" + productId;
        String productInfo = redisTemplate.opsForValue().get(key);

        if (productInfo == null) {
            // 第一次检查：缓存是否为空
            String lockKey = LOCK_PREFIX + productId;
            Boolean locked = redisTemplate.opsForValue().setIfAbsent(lockKey, "locked", 10, TimeUnit.SECONDS);

            if (locked != null && locked) {
                try {
                    // 第二次检查：加锁后，再次检查缓存是否为空，防止多个线程同时进入加锁块
                    productInfo = redisTemplate.opsForValue().get(key);
                    if(productInfo == null) {
                        productInfo = loadProductFromDatabase(productId);

                        if (productInfo != null) {
                            redisTemplate.opsForValue().set(key, productInfo, 3600, TimeUnit.SECONDS);
                        }
                    }

                } finally {
                    redisTemplate.delete(lockKey);
                }
            } else {
                // 没有获取到锁，等待一段时间后重试
                try {
                    Thread.sleep(50);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                return getProductInfoWithDoubleCheck(productId);
            }
        }

        return productInfo;
    }

    private String loadProductFromDatabase(String productId) {
        // 模拟从数据库加载数据
        try {
            Thread.sleep(100); // 模拟数据库查询耗时
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "Product Info for " + productId;
    }
}

说明: 在加锁之前，getProductInfoWithDoubleCheck方法先检查缓存是否为空。如果缓存为空，则尝试获取锁。获取到锁之后，再次检查缓存是否为空，这是为了防止多个线程同时进入加锁块。

2.4 采用永不过期的"逻辑过期"策略

不设置过期时间，在缓存数据中增加一个过期时间字段，由程序来判断缓存是否过期。可以使用后台线程定时刷新缓存。

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;

@Service
public class ProductService {

    @Autowired
    private StringRedisTemplate redisTemplate;

    @Autowired
    private ObjectMapper objectMapper;

    private static final long LOGICAL_EXPIRE_TIME = 3600; // 逻辑过期时间，1小时

    public String getProductInfoWithLogicalExpire(String productId) {
        String key = "product:" + productId;
        String productInfoJson = redisTemplate.opsForValue().get(key);

        if (productInfoJson == null) {
            // 缓存为空，直接加载
            return loadProductAndSetCache(productId);
        }

        try {
            Map<String, Object> cacheData = objectMapper.readValue(productInfoJson, HashMap.class);
            long expireTime = (long) cacheData.get("expireTime");
            String data = (String) cacheData.get("data");

            if (System.currentTimeMillis() > expireTime) {
                // 逻辑过期，开启一个线程异步刷新缓存
                new Thread(() -> {
                    loadProductAndSetCache(productId);
                }).start();
                return data; // 返回旧数据
            } else {
                // 缓存未过期，直接返回
                return data;
            }

        } catch (IOException e) {
            // JSON解析异常，重新加载
            return loadProductAndSetCache(productId);
        }
    }

    private String loadProductAndSetCache(String productId) {
        String productInfo = loadProductFromDatabase(productId);
        if (productInfo != null) {
            cacheProductInfo(productId, productInfo);
        }
        return productInfo;
    }

    private void cacheProductInfo(String productId, String productInfo) {
        String key = "product:" + productId;
        Map<String, Object> cacheData = new HashMap<>();
        cacheData.put("data", productInfo);
        cacheData.put("expireTime", System.currentTimeMillis() + LOGICAL_EXPIRE_TIME * 1000); // 设置逻辑过期时间

        try {
            String json = objectMapper.writeValueAsString(cacheData);
            redisTemplate.opsForValue().set(key, json);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
    }

    private String loadProductFromDatabase(String productId) {
        // 模拟从数据库加载数据
        try {
            Thread.sleep(100); // 模拟数据库查询耗时
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "Product Info for " + productId;
    }

    //模拟定时刷新
    @Scheduled(fixedRate = 60000) // 每分钟执行一次
    public void refreshCache(){
        // 实际场景中，这里应该查询需要刷新的key，然后进行刷新
        System.out.println("刷新缓存...");
    }

}

说明: getProductInfoWithLogicalExpire方法首先从缓存中获取数据，如果缓存为空，则直接加载。如果缓存存在，则判断是否逻辑过期。如果逻辑过期，则开启一个线程异步刷新缓存，并返回旧数据。

2.5 使用熔断器(Circuit Breaker)

熔断器是一种保护机制，当服务出现故障时，熔断器会阻止请求访问该服务，防止故障蔓延。可以使用Hystrix或Resilience4j等熔断器框架。

2.6 缓存预热

在系统启动时，预先加载一些热点数据到缓存中，避免在系统运行过程中，大量请求同时访问数据库。

3. 缓存服务高可用架构方案

除了上述策略，构建高可用的缓存服务也是预防缓存雪崩的关键。

3.1 Redis集群

使用Redis集群可以提高缓存服务的可用性和性能。Redis集群采用分片的方式存储数据，即使部分节点宕机，仍然可以提供服务。常用的Redis集群方案包括：

  • Redis Cluster: Redis官方提供的集群方案，支持自动分片和故障转移。
  • Codis: 豌豆荚开源的Redis集群方案，支持动态扩容和缩容。
  • Twemproxy: Twitter开源的Redis代理，支持分片和连接池。

3.2 多级缓存

使用多级缓存可以提高缓存的命中率和性能。常用的多级缓存架构包括：

  • 本地缓存 + 分布式缓存: 使用本地缓存（如Caffeine）作为一级缓存，分布式缓存（如Redis）作为二级缓存。本地缓存速度快，但容量有限；分布式缓存容量大，但速度相对较慢。
  • 多Region Redis: 将Redis部署在不同的Region，提高可用性

3.3 异地多活

为了保证在整个数据中心发生故障时，服务仍然可用，可以采用异地多活的架构。将应用和缓存服务部署在不同的数据中心，并进行数据同步。

4. 架构方案示例：Spring Boot + Redis Cluster + Caffeine

下面是一个结合Spring Boot、Redis Cluster和Caffeine的缓存架构方案示例：

// Caffeine配置
@Configuration
@EnableCaching
public class CaffeineConfig {

    @Bean
    public Caffeine<Object, Object> caffeineConfig() {
        return Caffeine.newBuilder()
                .initialCapacity(100)
                .maximumSize(1000)
                .expireAfterWrite(5, TimeUnit.MINUTES);
    }

    @Bean
    public CacheManager cacheManager(Caffeine<Object, Object> caffeine) {
        CaffeineCacheManager caffeineCacheManager = new CaffeineCacheManager();
        caffeineCacheManager.setCaffeine(caffeine);
        return caffeineCacheManager;
    }
}
// Redis配置
@Configuration
public class RedisConfig {

    @Bean
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) {
        RedisTemplate<String, Object> template = new RedisTemplate<>();
        template.setConnectionFactory(redisConnectionFactory);
        // 配置key的序列化方式
        template.setKeySerializer(new StringRedisSerializer());
        // 配置value的序列化方式，可以使用Jackson2JsonRedisSerializer或GenericJackson2JsonRedisSerializer
        template.setValueSerializer(new GenericJackson2JsonRedisSerializer());
        template.setHashKeySerializer(new StringRedisSerializer());
        template.setHashValueSerializer(new GenericJackson2JsonRedisSerializer());
        template.afterPropertiesSet();
        return template;
    }
}
import com.github.benmanes.caffeine.cache.Cache;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cache.CacheManager;
import org.springframework.cache.annotation.Cacheable;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import java.util.concurrent.TimeUnit;

@Service
public class ProductService {

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    @Autowired
    private CacheManager cacheManager;

    // 使用Caffeine作为本地缓存，Redis作为二级缓存
    @Cacheable(cacheNames = "productInfo", key = "#productId")
    public String getProductInfo(String productId) {
        String key = "product:" + productId;
        String productInfo = (String) redisTemplate.opsForValue().get(key);

        if (productInfo == null) {
            // 从数据库获取数据
            productInfo = loadProductFromDatabase(productId);

            if (productInfo != null) {
                // 设置缓存，过期时间随机化
                redisTemplate.opsForValue().set(key, productInfo, 3600, TimeUnit.SECONDS);
            }
        }
        return productInfo;
    }

    private String loadProductFromDatabase(String productId) {
        // 模拟从数据库加载数据
        try {
            Thread.sleep(100); // 模拟数据库查询耗时
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "Product Info for " + productId;
    }

    // 清空本地缓存
    public void clearLocalCache(String productId) {
        Cache cache = (Cache) cacheManager.getCache("productInfo").getNativeCache();
        cache.invalidate(productId);
    }
}

说明:

  • CaffeineConfig类配置Caffeine缓存，设置初始容量、最大容量和过期时间。
  • RedisConfig类配置Redis连接和序列化方式。
  • ProductService类使用@Cacheable注解将数据缓存到Caffeine中，如果Caffeine中不存在，则从Redis中获取数据，如果Redis中也不存在，则从数据库中获取数据，并同时更新Caffeine和Redis。
  • clearLocalCache方法用于清空本地缓存。

5. 监控与告警

对于缓存系统，完善的监控与告警机制至关重要。我们需要监控缓存的命中率、响应时间、错误率等指标，并在出现异常时及时发出告警。常用的监控工具包括：

  • Prometheus: 开源的监控系统，可以收集和存储各种指标数据。
  • Grafana: 开源的数据可视化工具，可以将Prometheus收集的指标数据以图表的形式展示出来。
  • RedisInsight: Redis官方提供的可视化工具，可以监控Redis的性能和状态。

6. 总结：多管齐下，保障系统稳定

总而言之，预防和解决Spring Boot应用中的缓存雪崩问题，需要综合运用多种策略和架构方案。从代码层面的过期时间分散化、互斥锁、双重检测锁、逻辑过期，到架构层面的Redis集群、多级缓存、异地多活，再到监控与告警，每一个环节都至关重要。只有构建一个完善的缓存体系，才能保障系统在高并发场景下的稳定性和可用性。

7. 一些建议

  • 选择合适的缓存策略: 根据业务场景选择合适的缓存策略。例如，对于高频访问的热点数据，可以使用互斥锁或双重检测锁；对于允许短暂不一致的数据，可以使用逻辑过期。
  • 合理设置缓存过期时间: 避免大量的key同时过期，可以将过期时间分散化。
  • 监控缓存性能: 监控缓存的命中率、响应时间、错误率等指标，及时发现和解决问题。
  • 进行压力测试: 在生产环境上线之前，进行充分的压力测试，评估系统的承载能力。

希望今天的分享对大家有所帮助，谢谢！

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注