浅谈redis缓存及缓存雪崩的处理

小鱼儿 2022-12-15 14:15 121阅读 0赞

**目录**

前言

代码分析

第一种代码案例：

第二种方案，加锁

第三种方案：semaphore实现共享锁

第四种方案：基于DCL（Double Check Lock）模式，结合Semaphore，再次进一步对代码进行优化。

第五种方案，进一步容错降级

--------------------

# 前言 #

现在随着redis应用的越来越广泛，以及高并发情况的出现，在大多数的springboot项目中，使用redis作为缓存，越来越普遍了，而伴随而来的，在项目中应用redis作为缓存，如何才能更好的使用，以及怎样避免雪崩，成为了项目架构越来越关心的事了。

*  缓存雪崩

先看看百度百科给的缓存雪崩的定义：

缓存雪崩是指缓存中数据大批量到过期时间，而查询数据量巨大，引起数据库压力过大甚至down机。

举例来说，在一个高并发的接口上，我们使用了redis缓存进行数据的查询，redis的缓存的性能假设在12W/qbs,而连接的mysql数据库的性能假设在5W/qbs(数据库的查询效率肯定要比redis低很多)，那么，正常情况下，没有任何问题。那么会有下面几种情况出现：

1.  网络故障（这种情况暂不考虑，在网络故障的情况下，应用大部分应该都访问不了了吧）
2.  redis重启（这种情况下，若没有进行持久化，数据势必会丢失，因为redis是内存数据库）
3.   缓存过期机制（例如对一些高并发的接口存储的数据，缓存过期机制设置的时间在接近 的时间点，那么可能会造成缓存MISS）
4.  内存不够用

针对上面几种情况：

1.  redis重启我们可以采用对应的持久化技术进行缓存数据持久化（RDB/AOF）
2.  缓存过期机制引起的问题，我们可以结合业务场景，将热数据的缓存过期时间不设置（缓存量小的情况下），或者对过期时间进行随机分散处理（不是很完善）。
3.  对于一些大数据量的热数据，可以在系统部署上线前进行预热处理（上线前，将数据记录批量的从数据库中存放到redis缓存中）
4.  对于内存不够用，就要结合业务的实际情况，在系统上线前进行充分的评估，提前对大数据量进行预估，或者分片集群存储。

**但是，总归一句话，缓存MISS是不可避免的，总会出现缓存MISS，那么我们怎么通过代码去控制呢？----加锁**

# **代码分析** #

**下面结合一个简单的查询用户代码案例，看几种处理情况，我的环境是SpringBoot+Mybatis+redis**

### **第一种代码案例：** ###

@Override
        public User findUserById(Integer id) {
    
            String cacheValue = (String) this.redisUtil.get(String.valueOf(id));
            if (null != cacheValue) {
                //先在redis 缓存库中查询，若缓存有值，则直接返回值
                System.out.println("缓存命中。。。。。。。");
                return JsonUtils.jsonToPojo(cacheValue,User.class);
            }
    
            // 后续逻辑，若缓存MISS异常情况出现之后，则去数据库中查询，然后重建缓存
            User user = this.userMapper.findUserById(id);
            if (null != user) {
                System.out.println("缓存MISS，查询数据库，重建缓存。。。");
                this.redisUtil.set(String.valueOf(id), JsonUtils.objectToJson(user));
                return user;
            }
    
            return null;
        }

解释：

如上代码，先在缓存中查询，若缓存命中，则直接返回数据，若出现缓存MISS的异常，则直接去数据库中查询，然后重建缓存，并返回数据。

弊端：

如上代码，在一般的小并发的情况下，是不会有什么大问题的，但是如果是高并发的接口的话，那么一旦在某一刻出现了缓存MISS，那么这个时候就都去访问数据库，就很容易使数据库扛不住压力，出现缓存雪崩。

那么我们就结合常用的方案，这个时候，就需要对缓存查询数据库，重建 缓存的时候进行加锁了，就可能会有如下改进代码：

### **第二种方案，加锁** ###

@Override
        public User findUserById(Integer id) {
    
            String cacheValue = (String) this.redisUtil.get(String.valueOf(id));
            if (null != cacheValue) {
                //先在redis 缓存库中查询，若缓存有值，则直接返回值
                System.out.println("缓存命中。。。。。。。");
                return JsonUtils.jsonToPojo(cacheValue,User.class);
            }
    
            // 后续逻辑，若缓存MISS异常情况出现之后，则去数据库中查询，然后重建缓存
            synchronized (this){
                //加同步锁关键字，进行加锁处理，那么访问到此代码块时，线程就需要排队等候了。。
                User user = this.userMapper.findUserById(id);
                if (null != user) {
                    System.out.println("缓存MISS，查询数据库，重建缓存。。。");
                    this.redisUtil.set(String.valueOf(id), JsonUtils.objectToJson(user));
                    return user;
                }
            }
    
            return null;
        }

解释：

通过加同步关键字很自然的实现加锁，但这种锁属于悲观锁，那么假如在某一刻，同时有12w的并发，那么不好意思，一次只能有一个人通过，其它人都在等候吧，显然，这种锁的机制对于一些追求实时性的网站来说，很不友好。当然，对于量不是那么大的来说，也是一种解决方案。

基于第二种方案，那么我们就想了，那么有没有一种办法，让一次多个人获得锁呢，而不是一次只能一个人拿到锁，即给代码块实现共享锁

### **第三种方案：semaphore实现共享锁** ###

解释：

上面代码主要使用了Semaphore的特性，实现指定多线程同步机制

Semaphore也是一个线程同步的辅助类，可以维护当前访问自身的线程个数，并提供了同步机制。使用Semaphore可以控制同时访问资源的线程个数，例如，实现一个文件允许的并发访问数。
    
    Semaphore的主要方法摘要：
        
    　　void acquire():从此信号量获取一个许可，在提供一个许可前一直将线程阻塞，否则线程被中断。
    
    　　void release():释放一个许可，将其返回给信号量。
    
    　　int availablePermits():返回此信号量中当前可用的许可数。
    
    　　boolean hasQueuedThreads():查询是否有线程正在等待获取。

这样的话，我们就从之前的一次一个人拿的独享锁，变成了一次多人（人数可以根据实际情况指定，例如上面指定了50人）的共享锁。

### **第四种方案：基于DCL（Double Check Lock）模式，结合Semaphore，再次进一步对代码进行优化。** ###

@Service
    public class UserServiceImpl implements UserService {
    
        @Autowired
        private UserMapper userMapper;
        @Autowired
        RedisUtil redisUtil;
    
         Semaphore semaphore = new Semaphore(50);//一次可以允许50个线程访问
       
    
        @Override
        public User findUserById(Integer id) {
    
            String cacheValue = (String) this.redisUtil.get(String.valueOf(id));
            if (null != cacheValue) {
                //先在redis 缓存库中查询，若缓存有值，则直接返回值
                System.out.println("缓存命中。。。。。。。");
                return JsonUtils.jsonToPojo(cacheValue,User.class);
            }
    
            //缓存MISS出现，则进行异常处理
    
            try {
                semaphore.acquire();
    
                //进入同步线程后，首先再次检查缓存是否存在，即DCL中的第二次检查
                cacheValue = (String) this.redisUtil.get(String.valueOf(id));
                if (null != cacheValue) {
                    //先在redis 缓存库中查询，若缓存有值，则直接返回值
                    System.out.println("缓存命中。。。。。。。");
                    return JsonUtils.jsonToPojo(cacheValue,User.class);
                }
    
                // 若缓存MISS异常情况出现之后，则去数据库中查询，然后重建缓存
                User user = this.userMapper.findUserById(id);
                if (null != user) {
                    System.out.println("缓存MISS，查询数据库，重建缓存。。。");
                    this.redisUtil.set(String.valueOf(id), JsonUtils.objectToJson(user));
                    return user;
                }
    
            } catch (InterruptedException e) {
                e.printStackTrace();
            }finally {
                semaphore.release();
            }
    
    
    
            return null;
        }
    }

解释：

上面的代码到尽头了吗？试想下，上面允许一次最大50个线程同步，那么假如对于高并发的，一次12W的并发量的话，那么50个线程意外的其他人访问，是不是就一直在等待中，这样显然还是不够友好，因此，我们会**容错降级，** 当出现等待的情况时，我们根据业务实际情况进行友好提示。

### **第五种方案，进一步容错降级** ###

@Service
    public class UserServiceImpl implements UserService {
    
        @Autowired
        private UserMapper userMapper;
        @Autowired
        RedisUtil redisUtil;
    
         Semaphore semaphore = new Semaphore(50);//一次可以允许50个线程访问
        
    
        @Override
        public User findUserById(Integer id) {
    
            String cacheValue = (String) this.redisUtil.get(String.valueOf(id));
            if (null != cacheValue) {
                //先在redis 缓存库中查询，若缓存有值，则直接返回值
                System.out.println("缓存命中。。。。。。。");
                return JsonUtils.jsonToPojo(cacheValue,User.class);
            }
    
            //缓存MISS出现，则进行异常处理
    
            try {
                boolean flag = semaphore.tryAcquire();
                if(!flag){
                    //没有得到许可
                    return new User();
                    //这里可以根据实际业务情况，对用户提示等待或稍后重试等信息，
                    // 而不至于让前端了浏览器一直在等待
                }
    
                //进入同步线程后，首先再次检查缓存是否存在，即DCL中的第二次检查
                cacheValue = (String) this.redisUtil.get(String.valueOf(id));
                if (null != cacheValue) {
                    //先在redis 缓存库中查询，若缓存有值，则直接返回值
                    System.out.println("缓存命中。。。。。。。");
                    return JsonUtils.jsonToPojo(cacheValue,User.class);
                }
    
                // 若缓存MISS异常情况出现之后，则去数据库中查询，然后重建缓存
                User user = this.userMapper.findUserById(id);
                if (null != user) {
                    System.out.println("缓存MISS，查询数据库，重建缓存。。。");
                    this.redisUtil.set(String.valueOf(id), JsonUtils.objectToJson(user));
                    return user;
                }
    
            } catch (Exception e) {
                e.printStackTrace();
            }finally {
                semaphore.release();
            }
    
    
    
            return null;
        }

无参方法tryAcquire（）的作用是尝试的获得1个许可，如果获取不到则返回false，
    该方法通常与if语句结合使用，其具有无阻塞的特点。
    无阻塞的特点可以使线程不至于在同步处一直持续等待的状态，
    如果if语句判断不成立则线程会继续走slse语句，程序会继续向下运行。

ok,至此，以上根据代码一步步进行了优化，当然，肯定还有更优的方案，毕竟技术无止境，这里只是阐述一种思路和理念。