緩存擊穿！竟然不知道怎么寫代碼？？？
在Redis中有三大問題：緩存雪崩、緩存擊穿、緩存穿透，今天我們來聊聊緩存擊穿。
關于緩存擊穿相關理論文章，相信大家已經看過不少，但是具體代碼中是怎么實現的，怎么解決的等問題，可能就一臉懵逼了。
今天，老田就帶大家來看看，緩存擊穿解決和代碼實現。
場景
請看下面這段代碼：
- /**
- * @author 田維常
- * @公眾號 java后端技術全棧
- * @date 2021/6/27 15:59
- */
- @Service
- public class UserInfoServiceImpl implements UserInfoService {
- @Resource
- private UserMapper userMapper;
- @Resource
- private RedisTemplate<Long, String> redisTemplate;
- @Override
- public UserInfo findById(Long id) {
- //查詢緩存
- String userInfoStr = redisTemplate.opsForValue().get(id);
- //如果緩存中不存在，查詢數據庫
- //1
- if (isEmpty(userInfoStr)) {
- UserInfo userInfo = userMapper.findById(id);
- //數據庫中不存在
- if(userInfo == null){
- return null;
- }
- userInfoStr = JSON.toJSONString(userInfo);
- //2
- //放入緩存
- redisTemplate.opsForValue().set(id, userInfoStr);
- }
- return JSON.parseObject(userInfoStr, UserInfo.class);
- }
- private boolean isEmpty(String string) {
- return !StringUtils.hasText(string);
- }
- }
整個流程：
如果，在//1到//2之間耗時1.5秒，那就代表著在這1.5秒時間內所有的查詢都會走查詢數據庫。這也就是我們所說的緩存中的“緩存擊穿”。
其實，你們項目如果并發量不是很高，也不用怕，并且我見過很多項目也就差不多是這么寫的，也沒那么多事，畢竟只是第一次的時候可能會發生緩存擊穿。
但，我們也不要抱著一個僥幸的心態去寫代碼，既然是多線程導致的，估計很多人會想到鎖，下面我們使用鎖來解決。
改進版
既然使用到鎖，那么我們第一時間應該關心的是鎖的粒度。
如果我們放在方法findById上，那就是所有查詢都會有鎖的競爭，這里我相信大家都知道我們為什么不放在方法上。
- /**
- * @author 田維常
- * @公眾號 java后端技術全棧
- * @date 2021/6/27 15:59
- */
- @Service
- public class UserInfoServiceImpl implements UserInfoService {
- @Resource
- private UserMapper userMapper;
- @Resource
- private RedisTemplate<Long, String> redisTemplate;
- @Override
- public UserInfo findById(Long id) {
- //查詢緩存
- String userInfoStr = redisTemplate.opsForValue().get(id);
- if (isEmpty(userInfoStr)) {
- //只有不存的情況存在鎖
- synchronized (UserInfoServiceImpl.class){
- UserInfo userInfo = userMapper.findById(id);
- //數據庫中不存在
- if(userInfo == null){
- return null;
- }
- userInfoStr = JSON.toJSONString(userInfo);
- //放入緩存
- redisTemplate.opsForValue().set(id, userInfoStr);
- }
- }
- return JSON.parseObject(userInfoStr, UserInfo.class);
- }
- private boolean isEmpty(String string) {
- return !StringUtils.hasText(string);
- }
- }
看似解決問題了，其實，問題還是沒得到解決，還是會緩存擊穿，因為排隊獲取到鎖后，還是會執行同步塊代碼，也就是還會查詢數據庫，完全沒有解決緩存擊穿。
雙重檢查鎖
由此，我們引入雙重檢查鎖，我們在上的版本中進行稍微改變，在同步模塊中再次校驗緩存中是否存在。
- /**
- * @author 田維常
- * @公眾號 java后端技術全棧
- * @date 2021/6/27 15:59
- */
- @Service
- public class UserInfoServiceImpl implements UserInfoService {
- @Resource
- private UserMapper userMapper;
- @Resource
- private RedisTemplate<Long, String> redisTemplate;
- @Override
- public UserInfo findById(Long id) {
- //查緩存
- String userInfoStr = redisTemplate.opsForValue().get(id);
- //第一次校驗緩存是否存在
- if (isEmpty(userInfoStr)) {
- //上鎖
- synchronized (UserInfoServiceImpl.class){
- //再次查詢緩存，目的是判斷是否前面的線程已經set過了
- userInfoStr = redisTemplate.opsForValue().get(id);
- //第二次校驗緩存是否存在
- if (isEmpty(userInfoStr)) {
- UserInfo userInfo = userMapper.findById(id);
- //數據庫中不存在
- if(userInfo == null){
- return null;
- }
- userInfoStr = JSON.toJSONString(userInfo);
- //放入緩存
- redisTemplate.opsForValue().set(id, userInfoStr);
- }
- }
- }
- return JSON.parseObject(userInfoStr, UserInfo.class);
- }
- private boolean isEmpty(String string) {
- return !StringUtils.hasText(string);
- }
- }
這樣，看起來我們就解決了緩存擊穿問題，大家覺得解決了嗎?
惡意攻擊
回顧上面的案例，在正常的情況下是沒問題，但是一旦有人惡意攻擊呢?
比如說：入參id=10000000，在數據庫里并沒有這個id，怎么辦呢?
第一步、緩存中不存在
第二步、查詢數據庫
第三步、由于數據庫中不存在，直接返回了，并沒有操作緩存
第四步、再次執行第一步.....死循環了吧
方案1：設置空對象
就是當緩存中和數據庫中都不存在的情況下，以id為key，空對象為value。
- set(id,空對象);
回到上面的四步，就變成了。
比如說：入參id=10000000，在數據庫里并沒有這個id，怎么辦呢?
第一步、緩存中不存在
第二步、查詢數據庫
第三步、由于數據庫中不存在，以id為key，空對象為value放入緩存中
第四步、執行第一步，此時，緩存就存在了，只是這時候只是一個空對象。
代碼實現部分：
- /**
- * @author 田維常
- * @公眾號 java后端技術全棧
- * @date 2021/6/27 15:59
- */
- @Service
- public class UserInfoServiceImpl implements UserInfoService {
- @Resource
- private UserMapper userMapper;
- @Resource
- private RedisTemplate<Long, String> redisTemplate;
- @Override
- public UserInfo findById(Long id) {
- String userInfoStr = redisTemplate.opsForValue().get(id);
- //判斷緩存是否存在，是否為空對象
- if (isEmpty(userInfoStr)) {
- synchronized (UserInfoServiceImpl.class){
- userInfoStr = redisTemplate.opsForValue().get(id);
- if (isEmpty(userInfoStr)) {
- UserInfo userInfo = userMapper.findById(id);
- if(userInfo == null){
- //構建一個空對象
- userInfo= new UserInfo();
- }
- userInfoStr = JSON.toJSONString(userInfo);
- redisTemplate.opsForValue().set(id, userInfoStr);
- }
- }
- }
- UserInfo userInfo = JSON.parseObject(userInfoStr, UserInfo.class);
- //空對象處理
- if(userInfo.getId() == null){
- return null;
- }
- return JSON.parseObject(userInfoStr, UserInfo.class);
- }
- private boolean isEmpty(String string) {
- return !StringUtils.hasText(string);
- }
- }
方案2 布隆過濾器
布隆過濾器(Bloom Filter):是一種空間效率極高的概率型算法和數據結構，用于判斷一個元素是否在集合中(類似Hashset)。它的核心一個很長的二進制向量和一系列hash函數，數組長度以及hash函數的個數都是動態確定的。
Hash函數：SHA1，SHA256，MD5..
布隆過濾器的用處就是，能夠迅速判斷一個元素是否在一個集合中。因此他有如下三個使用場景:
- 網頁爬蟲對URL的去重，避免爬取相同的URL地址
- 反垃圾郵件，從數十億個垃圾郵件列表中判斷某郵箱是否垃圾郵箱(垃圾短信)
- 緩存擊穿，將已存在的緩存放到布隆過濾器中，當黑客訪問不存在的緩存時迅速返回避免緩存及DB掛掉。
其內部維護一個全為0的bit數組，需要說明的是，布隆過濾器有一個誤判率的概念，誤判率越低，則數組越長，所占空間越大。誤判率越高則數組越小，所占的空間越小。布隆過濾器的相關理論和算法這里就不聊了，感興趣的可以自行研究。
優勢和劣勢
優勢
- 全量存儲但是不存儲元素本身，在某些對保密要求非常嚴格的場合有優勢;
- 空間高效率
- 插入/查詢時間都是常數O(k),遠遠超過一般的算法
劣勢
- 存在誤算率(False Positive),默認0.03，隨著存入的元素數量增加，誤算率隨之增加;
- 一般情況下不能從布隆過濾器中刪除元素;
- 數組長度以及hash函數個數確定過程復雜;
代碼實現：
- /**
- * @author 田維常
- * @公眾號 java后端技術全棧
- * @date 2021/6/27 15:59
- */
- @Service
- public class UserInfoServiceImpl implements UserInfoService {
- @Resource
- private UserMapper userMapper;
- @Resource
- private RedisTemplate<Long, String> redisTemplate;
- private static Long size = 1000000000L;
- private static BloomFilter<Long> bloomFilter = BloomFilter.create(Funnels.longFunnel(), size);
- @Override
- public UserInfo findById(Long id) {
- String userInfoStr = redisTemplate.opsForValue().get(id);
- if (isEmpty(userInfoStr)) {
- //校驗是否在布隆過濾器中
- if(bloomFilter.mightContain(id)){
- return null;
- }
- synchronized (UserInfoServiceImpl.class){
- userInfoStr = redisTemplate.opsForValue().get(id);
- if (isEmpty(userInfoStr) ) {
- if(bloomFilter.mightContain(id)){
- return null;
- }
- UserInfo userInfo = userMapper.findById(id);
- if(userInfo == null){
- //放入布隆過濾器中
- bloomFilter.put(id);
- return null;
- }
- userInfoStr = JSON.toJSONString(userInfo);
- redisTemplate.opsForValue().set(id, userInfoStr);
- }
- }
- }
- return JSON.parseObject(userInfoStr, UserInfo.class);
- }
- private boolean isEmpty(String string) {
- return !StringUtils.hasText(string);
- }
- }
方案3 互斥鎖
使用Redis實現分布式的時候，有用到setnx，這里大家可以想象，我們是否可以使用這個分布式鎖來解決緩存擊穿的問題?
這個方案留給大家去實現，只要掌握了Redis的分布式鎖，那這個實現起來就非常簡單了。
總結
搞定緩存擊穿、使用雙重檢查鎖的方式來解決，看到雙重檢查鎖，大家肯定第一印象就會想到單例模式，這里也算是給大家復習一把雙重檢查鎖的使用。
由于惡意攻擊導致的緩存擊穿，解決方案我們也實現了兩種，至少在工作和面試中，肯定是能應對了。
另外，使用鎖的時候注意鎖的力度，這里建議換成分布式鎖(Redis或者Zookeeper實現)，因為我們既然引入緩存，大部分情況下都會是部署多個節點的，同時，引入分布式鎖了，我們就可以使用方法入參id用起來，這樣是不是更爽!
希望大家能領悟到的是文中的一些思路，并不是死記硬背技術。
