缓存击穿、缓存穿透、缓存雪崩

￡神魔★判官ぃ 2022-03-12 14:14 343阅读 0赞

### 文章目录 ###

*  缓存击穿
 *  缓存穿透
 *  缓存雪崩
 *  双写一致性分析
 *   *  更新数据库，然后删除缓存
 *  布隆过滤器
 *   *  布隆过滤器使用案例

# 缓存击穿 #

*  当某一个缓存失效时候，瞬时并发或请求量很大大，由于缓存失效，直接穿透到数据库查询。
 *  解决办法：单机情况可以使用synchronized，分布式情况设置一个redis分布式锁（原生的是setnx对应redisson），注意使用锁的时候，最好加上超时时间。reentrantlock的定时锁tryLock(long, TimeUnit)或redis的超时时间

String get(String key) {  
       String value = redis.get(key);  
       if (value  == null) {  
        if (redis.setnx(key_mutex, "1")) {  
            // 3 min timeout to avoid mutex holder crash  
            redis.expire(key_mutex, 3 * 60)  
            value = db.get(key);  
            redis.set(key, value);  
            redis.delete(key_mutex);  
        } else {  
            //其他线程休息50毫秒后重试  
            Thread.sleep(50);  
            get(key);  
        }  
      }  
    }

# 缓存穿透 #

*  频繁的查询一个不存在的值，故意的攻击数据库
 *  解决办法：如果查询结果为空的时候，仍然把这个空结果进行缓存，但它的过期时间会很短，最长不超过五分钟，更新数据的时候，刷新缓存。方法二：采用布隆过滤器，将所有可能存在的数据哈希到一个足够大的bitmap中，一个一定不存在的数据会被 这个bitmap拦截掉，从而避免了对底层存储系统的查询压力。笔者认为采用过滤器的方法，首先要知道值的范围，才能针对性的设置bitmap。

# 缓存雪崩 #

*  多个缓存对象，设置的过期时间一致，在同一时间过期，大量查询数据库。
 *  解决办法：使用随机值的方式，设置缓存时间，比如50分钟到60分钟之间随机数

# 双写一致性分析 #

如何保证数据库和缓存中的数据一致？

## 更新数据库，然后删除缓存 ##

*  推荐方案，先更新数据库，然后删除缓存  
    此方案数据不一致的几率比较低，并且实现简单。  
    但是也有极端情况，造成数据不一致的情况：1.在缓存刚好失效时，有线程查询数据库得到旧值，另外一个线程更新数据库并删除缓存后，前面持有旧值的线程将数据存入缓存，造成数据不一致。2.也有可能，更新成功，删除失败。

**解决**：

1.  可以提供重试机制，过一段时间再次删除。
2.  使用数据库的binlog，数据库写入数据会有binlog写入，通过一个服务监听binlog，如果有数据变化，就在客户端删除缓存。  
    总之，删除缓存失败情况，就不断的重试删除。达到最终一致性。

*  不推荐先删缓存，再更新。  
    这种很容易导致数据不一致，  
    A要更新数据库，先删除了缓存（还未更新）。  
    这时，B来查询出来旧值，更新到了缓存（旧值）  
    最后，A更新了数据库。  
    这样就出现了ABA的问题，并发情况下，很容易不一致。  
    这种解决办法就是，写入数据之后，再次删除

# 布隆过滤器 #

经典面试题，如果在海量数据中（例如10亿无序，不定长，不重复）快速判断一个元素是否存在。

如何将元素和位图（位数组）下标对应

思路：  
1.将所有元素转换成相同长度  
2.如何做到均匀分布。

解决：使用hash算法。将所有的key，通过hash运算，放到位图的相应位置，将位置的状态标志为1。但是有可能，多个key通过hash算法之后，映射到同一个位置，这时就产生了hash碰撞。可以增大位图的长度，增加多次hash运算的方式，减少hash碰撞。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2E3MTg1MTUwMjg_size_16_color_FFFFFF_t_70]  
当设置了过滤器，我们在查询 key 的时候首先会以同样的方法计算出 key 的哈希值，然后在过滤器中从查找，这样 a 的哈希值 1 4 6 对应的位都是 1 ，表明这个 key 是可能存在的，可以查询缓存或者数据库，而 c 的哈希值 3 6 7 所对应的位为有两个是 0，表明这个 key 是绝对不存在的，这时就可以直接返回结果给客户端，避免了空值查询数据库。

## 布隆过滤器使用案例 ##

@RunWith(SpringJUnit4ClassRunner.class)
    @SpringBootTest
    @EnableAutoConfiguration
    public class BloomTestsConcurrency { 
        @Resource
        private RedisTemplate redisTemplate;
    
        @Autowired
        private UserService userService;
    
        private static final int THREAD_NUM = 1000; // 并发线程数量，Windows机器不要设置过大
    
        static BloomFilter<String> bf;
    
        static List<User> allUsers;
    
        @PostConstruct
        public void init() { 
            // 从数据库获取数据，加载到布隆过滤器
            long start = System.currentTimeMillis();
            // 从数据库查出来所有user，得到List<user>
            allUsers = userService.getAllUser();
            if (allUsers == null || allUsers.size() == 0) { 
                return;
            }
            // 创建布隆过滤器，默认误判率0.03，即3%
            bf = BloomFilter.create(Funnels.stringFunnel(Charsets.UTF_8), allUsers.size());
            // 误判率越低，数组长度越长，需要的哈希函数越多
            // bf = BloomFilter.create(Funnels.stringFunnel(Charsets.UTF_8), allUsers.size(), 0.0001);
            // 将数据存入布隆过滤器
            for (User user : allUsers) { 
                bf.put(user.getAccount());
            }
            long end = System.currentTimeMillis();
            System.out.println("查询并加载"+allUsers.size()+"条数据到布隆过滤器完毕，总耗时："+(end -start ) +"毫秒");
        }
    
        @Test
        public void cacheBreakDownTest() { 
            long start = System.currentTimeMillis();
            allUsers = userService.getAllUser();
            // 设置一个cyclicBarrier，指定线程数，
            CyclicBarrier cyclicBarrier = new CyclicBarrier(THREAD_NUM);
            ExecutorService executorService = Executors.newFixedThreadPool(THREAD_NUM);
            for (int i = 0; i < THREAD_NUM; i++){ 
           
           // 利用cyclicBarrier， MyThread中先用cyclicBarrier.await();阻塞住，当所有线程创建完成之后，再一起执行并发
                executorService.execute(new BloomTestsConcurrency().new MyThread(cyclicBarrier, redisTemplate, userService));
            }
    
            executorService.shutdown();
            //判断是否所有的线程已经运行完
            while (!executorService.isTerminated()) { 
    
            }
    
            long end = System.currentTimeMillis();
            System.out.println("并发数："+THREAD_NUM + "，新建线程以及过滤总耗时："+(end -start ) +"毫秒，演示结束");
        }
    
        public class MyThread implements Runnable { 
            private CyclicBarrier cyclicBarrier;
            private RedisTemplate redisTemplate;
            private UserService userService;
    
            public MyThread(CyclicBarrier cyclicBarrier, RedisTemplate redisTemplate, UserService userService) { 
                this.cyclicBarrier = cyclicBarrier;
                this.redisTemplate = redisTemplate;
                this.userService = userService;
            }
    
            @Override
            public void run() { 
                //所有子线程等待，当子线程全部创建完成再一起并发执行后面的代码
                try { 
                    cyclicBarrier.await();
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                } catch (BrokenBarrierException e) { 
                    e.printStackTrace();
                }
    
                // 1.1 （测试：布隆过滤器判断不存在，拦截——如果没有布隆过滤器，将造成缓存穿透）
                // 随机产生一个字符串，在redis中不存在
                String randomUser = UUID.randomUUID().toString();
                // 1.2 （测试：布隆过滤器判断存在，从Redis缓存取值，如果Redis为空则查询数据库并写入Redis）
                // 从List中获取一个存在的用户
                // String randomUser = allUsers.get(new Random().nextInt(allUsers.size())).getAccount();
                String key = "Key:" + randomUser;
    
                Date date1 = new Date();
                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    
                // 如果布隆过滤器中不存在这个用户直接返回，将流量挡掉
                if (!bf.mightContain(randomUser)) { 
                    System.out.println(sdf.format(date1)+" 布隆过滤器中不存在，非法请求");
                    return;
                }
    
                // 查询缓存，如果缓存中存在直接返回缓存数据
                ValueOperations<String, String> operation =
                        (ValueOperations<String, String>) redisTemplate.opsForValue();
                Object cacheUser = operation.get(key);
                if (cacheUser != null) { 
                    Date date2 = new Date();
                    System.out.println(sdf.format(date2)+" 命中redis缓存");
                    return;
                }
    
                // TODO 防止并发重复写缓存，加锁
                synchronized (randomUser) { 
                    // 如果缓存不存在查询数据库
                    List<User> user = userService.getUserByAccount(randomUser);
                    if (user == null || user.size() == 0) { 
                        // 很容易发生连接池不够用的情况 HikariPool-1 - Connection is not available, request timed out after
                        System.out.println(" Redis缓存不存在，查询数据库也不存在，发生缓存穿透！！！");
                        return;
                    }
                    // 将mysql数据库查询到的数据写入到redis中
                    Date date3 = new Date();
                    System.out.println(sdf.format(date3)+" 从数据库查询并写入Reids");
                    operation.set("Key:" + user.get(0).getAccount(), user.get(0).getAccount());
                }
            }
    
        }
    
    
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2E3MTg1MTUwMjg_size_16_color_FFFFFF_t_70]: /images/20220312/2208d175772f4424afe2e3036dea6159.png