在 Redis 存储的所有数据中，有一部分是被频繁访问的。有两种情况可能会导致热点问题的产生

一个是用户集中访问的数据，比如抢购的商品，明星结婚和明星出轨的微博

一种就是在数据进行分片的情况下，负载不均衡，超过了单个服务器的承受能力。热点问题可能引起缓存服务的不可用，最终造成压力堆积到数据库

热点数据发现

客户端

我们可以在所有调用了 get、set 方法的地方，加上 key 的计数。但是这样的话，每一个地方都要修改，重复的代码也多。如果我们用的是 Jedis 的客户端，我们可以在 Jedis 的 Connection 类的 sendCommand()里面，用一个 HashMap 进行 key 的计数

但是这种方式有几个问题：

1、不知道要存多少个 key，可能会发生内存泄露的问题 2、会对客户端的代码造成入侵 3、只能统计当前客户端的热点 key

代理层

在代理端实现，比如 TwemProxy 或者 Codis，但是不是所有的项目都使用了代理的架构

服务端

在服务端统计，Redis 有一个 monitor 的命令，可以监控到所有 Redis执行的命令

jedis.monitor(new JedisMonitor() { @Override public void onCommand(String command) { System.out.println("#monitor: " + command); } });

Facebook 的 开 源 项 目 redis-faina（https://github.com/facebookarchive/redis-faina.git）就是基于这个原理实现的。 它是一个 python 脚本，可以分析 monitor 的数据。

redis-cli -p 6379 monitor | head -n 100000 | ./redis-faina.py

这种方法也会有两个问题：

monitor 命令在高并发的场景下，会影响性能，所以不适合长时间使用只能统计一个 Redis 节点的热点 key

机器层面

还有一种方法就是机器层面的，通过对 TCP 协议进行抓包，也有一些开源的方案，比如 ELK 的 packetbeat 插件  

缓存雪崩

什么是缓存雪崩

缓存雪崩就是 Redis 的大量热点数据同时过期（失效），因为设置了相同的过期时间，刚好这个时候 Redis 请求的并发量又很大，就会导致所有的请求落到数据库

缓存雪崩 的解决方案

加互斥锁或者使用队列，针对同一个 key 只允许一个线程到数据库查询缓存定时预先更新，避免同时失效通过加随机数，使 key 在不同的时间过期缓存永不过期

 

缓存穿透

缓存穿透什么时候发生

数据在数据库和 Redis 里面都不存在，可能是一次条件错误的查询。在这种情况下，因为数据库值不存在，所以肯定不会写入 Redis，那么下一次查询相同的key 的时候，肯定还是会再到数据库查一次。

循环查询数据库中不存在的值，并且每次使用的是相同的 key 的情况，我们有没有什么办法避免应用到数据库查询呢？

缓存空数据缓存特殊字符串，比如&&

我们可以在数据库缓存一个空字符串，或者缓存一个特殊的字符串，那么在应用里面拿到这个特殊字符串的时候，就知道数据库没有值了，也没有必要再到数据库查询了

但是这里需要设置一个过期时间，不然的话数据库已经新增了这一条记录，应用也还是拿不到值

这个是应用重复查询同一个不存在的值的情况，如果应用每一次查询的不存在的值是不一样的呢？

即使你每次都缓存特殊字符串也没用，因为它的值不一样，比如我们的用户系统登录的场景，如果是恶意的请求，它每次都生成了一个符合 ID 规则的账号，但是这个账号在我们的数据库是不存在的，那 Redis 就完全失去了作用。

每次查询的值都不存在导致的 Redis 失效的情况，我们就把它叫做缓存穿透

 

经典面试题

如何在海量元素中（例如 10 亿无序、不定长、不重复）快速判断一个元素是否存在？

如果是缓存穿透的这个问题，我们要避免到数据库查询不存的数据，肯定要把这 10亿放在别的地方。这些数据在 Redis 里面也是没有的，为了加快检索速度，我们要把数据放到内存里面来判断，问题来了：

如果我们直接把这些元素的值放到基本的数据结构（List、Map、Tree）里面，比如一个元素 1 字节的字段，10 亿的数据大概需要 900G 的内存空间，这个对于普通的服务器来说是承受不了的。

我们存储这几十亿个元素，不能直接存值，我们应该找到一种最简单的最节省空间的数据结构，用来标记这个元素有没有出现

这个东西我们就把它叫做位图，他是一个有序的数组，只有两个值，0 和 1。0 代表不存在，1 代表存在

我们怎么用这个数组里面的有序的位置来标记这10亿个元素是否存在呢？是不是必须要有一个映射方法，把元素映射到一个下标位置上？

对于这个映射方法，有几个基本的要求：

1、因为我们的值长度是不固定的，我希望不同长度的输入，可以得到固定长度的输出。

2、转换成下标的时候，我希望他在我的这个有序数组里面是分布均匀的，不然的话全部挤到一对去了，我也没法判断到底哪个元素存了，哪个元素没存。

这个就是哈希函数，比如 MD5、SHA-1 等等这些都是常见的哈希算法

比如，这 4 个元素，我们经过哈希函数和位运算，得到了相应的下标

 

哈希碰撞

lily 和 tony 经过计算得到的哈希值是一样的，那么再经过位运算得到的下标肯定是一样的，我们把这种情况叫做哈希冲突或者哈希碰撞

如果发生了哈希碰撞，这个时候对于我们的容器存值肯定是有影响的，我们可以通过哪些方式去降低哈希碰撞的概率呢？

扩大维数组的长度或者说位图容量。因为我们的函数是分布均匀的，所以，位图容量越大，在同一个位置发生哈希碰撞的概率就越小如果两个元素经过一次哈希计算，得到的相同下标的概率比较高，可以计算多次， 原来我只用一个哈希函数，现在我对于每一个要存储的元素都用多个哈希函数计算，这样每次计算出来的下标都相同的概率就小得多了

 

布隆过滤器原理

布隆过滤器的本质就是一个位数组，和若干个哈希函数

集合里面有 3 个元素，要把它存到布隆过滤器里面去，应该怎么做？

首先是 a 元素，这里我们用 3 次计算。b、c 元素也一样

元素已经存进去之后，现在我要来判断一个元素在这个容器里面是否存在，就要使用同样的三个函数进行计算

比如 d 元素，我用第一个函数 f1 计算，发现这个位置上是 1，没问题。第二个位置也是 1，第三个位置也是 1 

如果经过三次计算得到的下标位置值都是 1，这种情况下，能不能确定 d 元素一定在这个容器里面呢？

实际上是不能的。比如这张图里面，这三个位置分别是把 a,b,c 存进去的时候置成 1 的，所以即使 d 元素之前没有存进去，也会得到三个 1，判断返回 true

这个是布隆过滤器的一个很重要的特性，因为哈希碰撞不可避免，所以它会存在一定的误判率。这种把本来不存在布隆过滤器中的元素误判为存在的情况，我们把它叫做假阳性（False Positive Probability，FPP）

布隆过滤器的特点

从容器的角度来说

1、如果布隆过滤器判断元素在集合中存在，不一定存在 2、如果布隆过滤器判断不存在，一定不存在

从元素的角度来说

3、如果元素实际存在，布隆过滤器一定判断存在 4、如果元素实际不存在，布隆过滤器可能判断存在

Guava的实现

谷歌的 Guava 里面就提供了一个现成的布隆过滤器

<dependency>     <groupId>com.google.guava</groupId>     <artifactId>guava</artifactId>     <version>21.0</version> </dependency>

创建布隆过滤器

BloomFilter<String> bf = BloomFilter.create(Funnels.stringFunnel(Charsets.UTF_8), insertions);

布隆过滤器提供的存放元素的方法是 put()

布隆过滤器提供的判断元素是否存在的方法是 mightContain()

if (bf.mightContain(data)) { if (sets.contains(data)) { // 判断存在实际存在的时候，命中 right++; continue; } // 判断存在却不存在的时候，错误 wrong++; }

布隆过滤器把误判率默认设置为 0.03，也可以在创建的时候指定

public static <T> BloomFilter<T> create(Funnel<? super T> funnel, long expectedInsertions) { return create(funnel, expectedInsertions, 0.03D); }

位图的容量是基于元素个数和误判率计算出来的

long numBits = optimalNumOfBits(expectedInsertions, fpp);

根据位数组的大小，我们进一步计算出了哈希函数的个数

int numHashFunctions = optimalNumOfHashFunctions(expectedInsertions, numBits);

存储 100 万个元素只占用了 0.87M 的内存，生成了 5 个哈希函数

演示地址

 

布隆过滤器 在项目中的使用

布隆过滤器的工作位置

因为要判断数据库的值是否存在，所以第一步是加载数据库所有的数据。在去 Redis查询之前，先在布隆过滤器查询，如果 bf 说没有，那数据库肯定没有，也不用去查了。如果 bf 说有，才走之前的流程

@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(); 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 = new CyclicBarrier(THREAD_NUM); ExecutorService executorService = Executors.newFixedThreadPool(THREAD_NUM); for (int i = 0; i < THREAD_NUM; i++){ 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 （测试：布隆过滤器判断不存在，拦截——如果没有布隆过滤器，将造成缓存穿透） // 随机产生一个字符串，在布隆过滤器中不存在 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()); } } } }

布隆过滤器 的其他应用场景

布隆过滤器解决的问题是什么？

如何在海量元素中快速判断一个元素是否存在。所以除了解决缓存穿透的问题之外，我们还有很多其他的用途

比如爬数据的爬虫，爬过的 url 我们不需要重复爬，那么在几十亿的 url 里面，怎么判断一个 url 是不是已经爬过了？比如我们的邮箱服务器，发送垃圾邮件的账号我们把它们叫做 spamer，在这么多的邮箱账号里面，怎么判断一个账号是不是 spamer 等等一些场景，我们都可以用到布隆过滤器