问题背景

我们都知道，在某种突发大流量的情景下，如果访问数据库的线程过多，会直接让数据库宕机。

这是我们可能会想到：Redis 的性能那么高，用 Redis 分担一下数据库的压力不久好了吗？

当然，这种办法确实为一种不错的思路，我们可以看一下这种思路的伪代码：

public Data getData(String name){
	Data data = 从Redis获取数据;
    //在并发环境下，多个线程可能同时判断数据在Redis中不存在，从而去执行从数据库获取数据的逻辑
	if(data == null){
		data = 从数据库获取数据；
		将数据写入Redis;
	}
	return data;
}

可以看出，这部分的代码其实存在一个严重的问题：虽然加了缓存，但是在高并发下很多线程还是会去访问数据库。

这就是我们常说的 缓存击穿。

一句话说清缓存击穿

某个“热点数据”在缓存中失效的瞬间，大量请求同时打到数据库。

缓存击穿的解决方案

逻辑过期

如果说缓存击穿的原因Key过期，那么不让Key过期不就行了。所以，我们可以设置Key的实际过期时间为No Limit(不过期)，在逻辑上给他设置一个过期时间。当访问这个数据时，对过期时间进行一个判断，如果过期了(设置时间 + 过期时间 < 当前时间)，那么先返回旧数据再异步重建缓存**。

**优点：**请求都不会越过Redis，所以响应速度快。重建缓存不会阻塞其他线程。
**缺点：**会导致短时间内的数据不一致问题。存的不过期Key越来越多，会导致占有Redis内存越来越多，需要设置清理策略。

分布式锁

当缓存失效时，不是立即去加载数据库数据，而是先使用分布式锁获取加载数据的权限，再去加载数据到缓存。这样，在第一个请求去加载数据时，其他并发请求则需要等待。第一个请求将数据加载到缓存后，直接释放分布式锁，此时其他请求就能够从缓存中获取数据，而不需要再去查询数据库。

**优点：**加锁保证只有一个线程能访问数据库，安全性高。重建缓存串行化，保证了数据一致性。
**缺点：**由于要等待第一个请求建立缓存，其他请求都会阻塞等待，响应速度变慢。由于使用到了分布式锁，实现复杂度高(需要处理锁的过期、重试、等待、死锁这些情况)。

在下文中，我们来看看用分布式锁解决这个问题。

分布式锁的实现

我们可以看看最简单的分布式锁流程：

实现代码：

public String getData(String name){
    String cachedValue = redisTemplate.opsForValue().get(name);
    if (cachedValue == null) {
        //分布式锁
        RLock lock = serviceLockTool.getLock(LockType.Reentrant, id);
        lock.lock();
        try {
            Data data = dataMapper.selectByname(name);
            if (data != null) {
                redisTemplate.opsForValue().set(name,JSON.toJSONString(data));
                cachedValue = JSON.toJSONString(data);
            }
        } finally {
            lock.unlock();
        }
    }
    return cachedValue;
}

我们可以看到，这段代码里每个请求确实都是排队执行了：尝试加锁 → 成功拿锁 → 查数据库 → 写入缓存 → 返回数据 → 释放锁。

但我们也会发现一个资源浪费的问题：明明第一个线程去重建缓存就够了，但是后面的线程还是会排队去重建缓存，这就造成资源浪费了。

所以我们就要实现一个效果：第一个请求从数据库查询出来放入缓存后，之后的请求都应该从缓存中查询。

下面便进入我们的重头戏：双重检测锁。

双重检测锁

在我们的Spring里，有一个单例模式，而单例Bean的创建，就用到了双重检测锁(Double-Check-Lock)。

下面，我们可以看一看源码。

protected Object getSingleton(String beanName, boolean allowEarlyReference) {
    // Quick check for existing instance without full singleton lock
    Object singletonObject = this.singletonObjects.get(beanName);
    //在未加锁时检测Bean是否存在
    if (singletonObject == null && isSingletonCurrentlyInCreation(beanName)) {
        singletonObject = this.earlySingletonObjects.get(beanName);
        if (singletonObject == null && allowEarlyReference) {
            // 此处开始加锁
            synchronized (this.singletonObjects) {
                // Consistent creation of early reference within full singleton lock
                // 这里在加锁后再检测一次Bean是否存在
                singletonObject = this.singletonObjects.get(beanName);
                if (singletonObject == null) {
                    singletonObject = this.earlySingletonObjects.get(beanName);
                    if (singletonObject == null) {
                        ObjectFactory<?> singletonFactory = this.singletonFactories.get(beanName);
                        if (singletonFactory != null) {
                            singletonObject = singletonFactory.getObject();
                            this.earlySingletonObjects.put(beanName, singletonObject);
                            this.singletonFactories.remove(beanName);
                        }
                    }
                }
            }
        }
    }
    return singletonObject;
    }

可以看到，双重检测锁的逻辑就是：先判断对象是否为null，如果为null的话，则加锁，在锁的逻辑中再判断一次对象是否为null，如果还是null，则进行创建对象。

我们在下面就用这个逻辑来优化代码，减少资源浪费。

分布式锁+双重检测锁

public String getData(String name){
	String cachedValue = redisTemplate.opsForValue().get(name);
    if (cachedValue == null){
        RLock lock = serviceLockTool.getLock(LockType.Reentrant, id);
        lock.lock();
        try{
            //这里检查是否已经有缓存
            cachedValue = redisTemplate.opsForValue().get(name);
            if (cachedValue == null){
                //两次从缓存查询都失败了再查数据库
                Data data = dataMapper.selectByName(name);
                if (data != null){
                    redisTemplate.opsForValue().set(name,JSON.toJSONString(data));
                    cachedValue = JSON.toJSONString(data);
                }
            }
        }
        finally{
            lock.unlock();
        }
    }
    return cachedValue;
}

获得锁后，先去缓存中查询，如果存在就直接将数据返回，如果还是不存在再从数据库查询然后放到缓存中。

这样我们就实现了第一个请求把数据库中数据放入缓存后，之后的请求直接从缓存中读取的效果。

那么这样就完美了吗？当然没有，这里还存在一个性能问题：第一个请求重建好缓存后，后续的请求不应该再去竞争锁，而是直接从缓存中获取数据并返回。

那么我们该怎么解决无用的锁竞争呢？我们可以采用tryLock方案：没抢到锁的线程等待一段时间，然后再去看看缓存里有没有数据。

减少锁竞争

public String getData(String name) {
    // 1.第一次查询缓存
    String cachedValue = redisTemplate.opsForValue().get(name);
    if (cachedValue == null) {
        RLock lock = serviceLockTool.getLock(LockType.Reentrant, "lock:" + name);
        boolean isLock = false;
        try {
            // 尝试加锁，等待1秒
            isLock = lock.tryLock(1, 10, TimeUnit.SECONDS); 
            if (isLock) {
            	//逻辑同上
            } else {
                // 2.加锁失败后的处理
                // 方案A: 睡100ms后递归重试
                Thread.sleep(100);
                return getData(name); 
                
                // 方案B: 再次尝试读缓存
                // return redisTemplate.opsForValue().get(name);
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException("系统繁忙，请稍后再试");
        } finally {
            // 安全释放锁：必须是自己持有的锁才能释放
            if (isLock && lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }
    }
    return cachedValue;
}

现在的逻辑就变成了：

第一个请求获得了锁，从数据库中查询再放入到缓存中
其余的请求等待1s来获得锁，如果超过1s，就不再竞争锁，程序继续往下执行获取锁失败的逻辑。由于第一个请求可能已经将放数据放入到缓存中了，所以可以直接从缓存中查询到，将数据返回

当然，即使是这样也还是会存在一些问题：如果第一个请求超过设定的等待时间，其余线程还是会执行加锁失败逻辑，导致一些其他问题。

总结

今天我们认识了缓存带来的派生问题：缓存击穿。

着重介绍了**双重检查锁：**请求进入时检查缓存是否存在，加锁后在检查一次。

并在过程中不断完善双重检查锁。当然，双重检查锁也并不完美，还是会在极端情况下出现问题。

但没有绝对完美的方案，使用缓存的高性能，就必须接受它带来的问题。在业务设计时，最重要的还是根据实际情况取舍，达到性能与安全的平衡。