There are three major problems in Redis: Cache avalanche, Cache breakdown, Cache Penetration, today we will talk about Cache Penetration.

I believe you have read a lot of theoretical articles related to cache breakdown, but you may be confused about how to implement it in the specific code and how to solve it.

Today, Lao Tian will take you to see the cache breakdown solution and code implementation.

Scenario

Please look at the following code:

/** 
 * @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);
        //如果緩存中不存在，查詢數(shù)據(jù)庫
        //1
        if (isEmpty(userInfoStr)) {
            UserInfo userInfo = userMapper.findById(id);
            //數(shù)據(jù)庫中不存在
            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);
    }
}

The entire process:

如果，在//1到//2之間耗時1.5秒，那就代表著在這1.5秒時間內所有的查詢都會走查詢數(shù)據(jù)庫。這也就是我們所說的緩存中的“緩存擊穿”。

其實，你們項目如果并發(fā)量不是很高，也不用怕，并且我見過很多項目也就差不多是這么寫的，也沒那么多事，畢竟只是第一次的時候可能會發(fā)生緩存擊穿。

但，我們也不要抱著一個僥幸的心態(tài)去寫代碼，既然是多線程導致的，估計很多人會想到鎖，下面我們使用鎖來解決。

改進版

既然使用到鎖，那么我們第一時間應該關心的是鎖的粒度。

如果我們放在方法findById上，那就是所有查詢都會有鎖的競爭，這里我相信大家都知道我們?yōu)槭裁床环旁诜椒ㄉ稀?/p>

/** 
 * @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);
                //數(shù)據(jù)庫中不存在
                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);
    }
}

看似解決問題了，其實，問題還是沒得到解決，還是會緩存擊穿，因為排隊獲取到鎖后，還是會執(zhí)行同步塊代碼，也就是還會查詢數(shù)據(jù)庫，完全沒有解決緩存擊穿。

雙重檢查鎖

由此，我們引入雙重檢查鎖，我們在上的版本中進行稍微改變，在同步模塊中再次校驗緩存中是否存在。

/** 
 * @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){ 
                //再次查詢緩存，目的是判斷是否前面的線程已經(jīng)set過了
                userInfoStr = redisTemplate.opsForValue().get(id);
                //第二次校驗緩存是否存在
                if (isEmpty(userInfoStr)) {
                    UserInfo userInfo = userMapper.findById(id);
                    //數(shù)據(jù)庫中不存在
                    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);
    }
}

這樣，看起來我們就解決了緩存擊穿問題，大家覺得解決了嗎？

惡意攻擊

回顧上面的案例，在正常的情況下是沒問題，但是一旦有人惡意攻擊呢？

比如說：入?yún)d=10000000，在數(shù)據(jù)庫里并沒有這個id，怎么辦呢？

第一步、緩存中不存在

第二步、查詢數(shù)據(jù)庫

第三步、由于數(shù)據(jù)庫中不存在，直接返回了，并沒有操作緩存

第四步、再次執(zhí)行第一步.....死循環(huán)了吧

方案1：設置空對象

就是當緩存中和數(shù)據(jù)庫中都不存在的情況下，以id為key，空對象為value。

set(id,空對象);

回到上面的四步，就變成了。

比如說：入?yún)?code style="font-size: 14px;padding: 2px 4px;border-radius: 4px;margin-right: 2px;margin-left: 2px;background-color: rgba(27, 31, 35, 0.05);font-family: "Operator Mono", Consolas, Monaco, Menlo, monospace;word-break: break-all;color: rgb(239, 112, 96);">id=10000000，在數(shù)據(jù)庫里并沒有這個id，怎么辦呢？

第一步、緩存中不存在

第二步、查詢數(shù)據(jù)庫

第三步、由于數(shù)據(jù)庫中不存在，以id為key，空對象為value放入緩存中

第四步、執(zhí)行第一步，此時，緩存就存在了，只是這時候只是一個空對象。

代碼實現(xiàn)部分：

/** 
 * @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):是一種空間效率極高的概率型算法和數(shù)據(jù)結構，用于判斷一個元素是否在集合中（類似Hashset）。它的核心一個很長的二進制向量和一系列hash函數(shù)，數(shù)組長度以及hash函數(shù)的個數(shù)都是動態(tài)確定的。

Hash函數(shù)：SHA1，SHA256，MD5..

布隆過濾器的用處就是，能夠迅速判斷一個元素是否在一個集合中。因此他有如下三個使用場景:

• Web crawler deduplicates URL to avoid crawling the same URL address
• Anti-spam Mail, determine whether a mailbox is spam (spam SMS) from billions of spam lists
• Cache breakdown, put the existing cache into the Bloom filter, When a hacker accesses a non-existent cache, return quickly to avoid cache and DB hang-ups.

It internally maintains a bit array that is all 0. It should be noted that the Bloom filter has a concept of false positive rate. The higher the false positive rate, the higher the false positive rate. If it is low, the longer the array is, the more space it takes up. The higher the false positive rate, the smaller the array and the smaller space it occupies. The relevant theories and algorithms of Bloom filters will not be discussed here. Those who are interested can study it by themselves.

Advantages and Disadvantages

Advantages

• Full storage but not The storage element itself has advantages in some situations where confidentiality requirements are very strict;
• High space efficiency
• Insertion/query time is ConstantO(k), far more than the general algorithm

Disadvantages

• 存在誤算率(False Positive),默認0.03，隨著存入的元素數(shù)量增加，誤算率隨之增加；
• 一般情況下不能從布隆過濾器中刪除元素；
• 數(shù)組長度以及hash函數(shù)個數(shù)確定過程復雜；

代碼實現(xiàn)：

/** 
 * @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實現(xiàn)分布式的時候，有用到setnx，這里大家可以想象，我們是否可以使用這個分布式鎖來解決緩存擊穿的問題？

這個方案留給大家去實現(xiàn)，只要掌握了Redis的分布式鎖，那這個實現(xiàn)起來就非常簡單了。

總結

搞定緩存擊穿、使用雙重檢查鎖的方式來解決，看到雙重檢查鎖，大家肯定第一印象就會想到單例模式，這里也算是給大家復習一把雙重檢查鎖的使用。

Due to cache breakdown caused by malicious attacks, we have also implemented two solutions. At least at work and interviews, we can definitely deal with it.

In addition, when using locks, pay attention to the lock strength. It is recommended to replace it with distributed lock(Redisor ZookeeperImplementation), because since we have introduced cache, in most cases we will deploy multiple nodes, and at the same time, introduced distributed locks, we can use the method Input parameter id to use, this is Isn’t it more exciting!

I hope everyone can understand some of the ideas in the article and not memorize the technology by rote.