緩存問題

我們熟知的是用到緩存就會遇到緩存三大問題：

• 緩存穿透
• 緩存擊穿
• 緩存雪崩

接下來讓我介紹在黑馬點評中這三個問題是如何解決了。

緩存穿透問題的解決思路

緩存穿透 ：緩存穿透是指客戶端請求的數據在緩存中和數據庫中都不存在，這樣緩存永遠不會生效，這些請求都會打到數據庫。

常見的解決方案有兩種：

• 緩存空對象
  • 優點：實現簡單，維護方便
  • 缺點：
    • 額外的內存消耗
    • 可能造成短期的不一致
• 布隆過濾
  • 優點：內存占用較少，沒有多余key
  • 缺點：
    • 實現復雜
    • 存在誤判可能

緩存空對象思路分析：當我們客戶端訪問不存在的數據時，先請求redis，但是此時redis中沒有數據，此時會訪問到數據庫，但是數據庫中也沒有數據，這個數據穿透了緩存，直擊數據庫，我們都知道數據庫能夠承載的并發不如redis這么高，如果大量的請求同時過來訪問這種不存在的數據，這些請求就都會訪問到數據庫，簡單的解決方案就是哪怕這個數據在數據庫中也不存在，我們也把這個數據存入到redis中去，這樣，下次用戶過來訪問這個不存在的數據，那么在redis中也能找到這個數據就不會進入到緩存了

布隆過濾：布隆過濾器其實采用的是哈希思想來解決這個問題，通過一個龐大的二進制數組，走哈希思想去判斷當前這個要查詢的這個數據是否存在，如果布隆過濾器判斷存在，則放行，這個請求會去訪問redis，哪怕此時redis中的數據過期了，但是數據庫中一定存在這個數據，在數據庫中查詢出來這個數據后，再將其放入到redis中，假設布隆過濾器判斷這個數據不存在，則直接返回。

這種方式優點在于節約內存空間，存在誤判，誤判原因在于：布隆過濾器走的是哈希思想，只要哈希思想，就可能存在哈希沖突

編碼解決商品查詢的緩存穿透問題

查詢數據時，如果這個數據不存在，還是會把這個數據寫入到Redis中，并且將value設置為空，歐當再次發起查詢時，我們如果發現命中之后，判斷這個value是否是null，如果是null，則是之前寫入的數據，證明是緩存穿透數據，如果不是，則直接返回數據。

public <R,ID> R queryWithPassThrough(String keyPrefix, ID id, Class<R> type, Function<ID, R> dbFallback, Long time, TimeUnit unit){String key = keyPrefix + id;// 1.從redis查詢商鋪緩存String json = stringRedisTemplate.opsForValue().get(key);// 2.判斷是否存在if (StrUtil.isNotBlank(json)) {// 3.存在，直接返回return JSONUtil.toBean(json, type);}// 判斷命中的是否是空值if (json != null) {// 返回一個錯誤信息return null;}// 4.不存在，根據id查詢數據庫R r = dbFallback.apply(id);// 5.不存在，返回錯誤if (r == null) {// 將空值寫入redisstringRedisTemplate.opsForValue().set(key, "", CACHE_NULL_TTL, TimeUnit.MINUTES);// 返回錯誤信息return null;}// 6.存在，寫入redisthis.set(key, r, time, unit);return r;}

緩存雪崩問題及解決思路

緩存雪崩是指在同一時段大量的緩存key同時失效或者Redis服務宕機，導致大量請求到達數據庫，帶來巨大壓力。

解決方案：

• 給不同的Key的TTL添加隨機值
• 利用Redis集群提高服務的可用性
• 給緩存業務添加降級限流策略
• 給業務添加多級緩存
  
  一般情況下，我們都采用隨機值過期的來解決雪崩問題。

緩存擊穿問題及解決思路

緩存擊穿問題也叫熱點Key問題，就是一個被高并發訪問并且緩存重建業務較復雜的key突然失效了，無數的請求訪問會在瞬間給數據庫帶來巨大的沖擊。

常見的解決方案有兩種：

互斥鎖方案：由于保證了互斥性，所以數據一致，且實現簡單，因為僅僅只需要加一把鎖而已，也沒其他的事情需要操心，所以沒有額外的內存消耗，缺點在于有鎖就有死鎖問題的發生，且只能串行執行性能肯定受到影響

邏輯過期方案 ：線程讀取過程中不需要等待，性能好，有一個額外的線程持有鎖去進行重構數據，但是在重構數據完成前，其他的線程只能返回之前的數據，且實現起來麻煩

問題分析

假設線程1在查詢緩存之后，本來應該去查詢數據庫，然后把這個數據重新加載到緩存的，此時只要線程1走完這個邏輯，其他線程就都能從緩存中加載這些數據了，但是假設在線程1沒有走完的時候，后續的線程2，線程3，線程4同時過來訪問當前這個方法， 那么這些線程都不能從緩存中查詢到數據，那么他們就會同一時刻來訪問查詢緩存，都沒查到，接著同一時間去訪問數據庫，同時的去執行數據庫代碼，對數據庫訪問壓力過大，如果有人利用此漏洞進行攻擊就會導致數據庫崩潰。

使用鎖來解決

因為鎖能實現互斥性。假設線程過來，只能一個人一個人的來訪問數據庫，從而避免對于數據庫訪問壓力過大，但這也會影響查詢的性能，因為此時會讓查詢的性能從并行變成了串行，我們可以采用tryLock方法 + double check來解決這樣的問題。

假設現在線程1過來訪問，他查詢緩存沒有命中，但是此時他獲得到了鎖的資源，那么線程1就會一個人去執行邏輯，假設現在線程2過來，線程2在執行過程中，并沒有獲得到鎖，那么線程2就可以進行到休眠，直到線程1把鎖釋放后，線程2獲得到鎖，然后再來執行邏輯，此時就能夠從緩存中拿到數據了。

代碼實現

加鎖的核心方案就是在redis未查詢到數據時，如果向數據庫就要獲取鎖，未獲取到鎖的線程就暫停執行，這樣就可以把原來并行的程序變成單線程的程序，從而減少并發帶來的問題。

public Shop queryWithMutex(Long id)  {String key = CACHE_SHOP_KEY + id;// 1、從redis中查詢商鋪緩存String shopJson = stringRedisTemplate.opsForValue().get("key");// 2、判斷是否存在if (StrUtil.isNotBlank(shopJson)) {// 存在,直接返回return JSONUtil.toBean(shopJson, Shop.class);}//判斷命中的值是否是空值if (shopJson != null) {//返回一個錯誤信息return null;}// 4.實現緩存重構//4.1 獲取互斥鎖String lockKey = "lock:shop:" + id;Shop shop = null;try {boolean isLock = tryLock(lockKey);// 4.2 判斷否獲取成功if(!isLock){//4.3 失敗，則休眠重試Thread.sleep(50);return queryWithMutex(id);}//4.4 成功，根據id查詢數據庫shop = getById(id);// 5.不存在，返回錯誤if(shop == null){//將空值寫入redisstringRedisTemplate.opsForValue().set(key,"",CACHE_NULL_TTL,TimeUnit.MINUTES);//返回錯誤信息return null;}//6.寫入redisstringRedisTemplate.opsForValue().set(key,JSONUtil.toJsonStr(shop),CACHE_NULL_TTL,TimeUnit.MINUTES);}catch (Exception e){throw new RuntimeException(e);}finally {//7.釋放互斥鎖unlock(lockKey);}return shop;}

邏輯過期方案

方案分析：我們之所以會出現這個緩存擊穿問題，主要原因是在于我們對key設置了過期時間，假設我們不設置過期時間，其實就不會有緩存擊穿的問題，但是不設置過期時間，這樣數據不就一直占用我們內存了嗎，我們可以采用邏輯過期方案。

我們把過期時間設置在 redis的value中，注意：這個過期時間并不會直接作用于redis，而是我們后續通過邏輯去處理。假設線程1去查詢緩存，然后從value中判斷出來當前的數據已經過期了，此時線程1去獲得互斥鎖，那么其他線程會進行阻塞，獲得了鎖的線程他會開啟一個 線程去進行 以前的重構數據的邏輯，直到新開的線程完成這個邏輯后，才釋放鎖， 而線程1直接進行返回，假設現在線程3過來訪問，由于線程線程2持有著鎖，所以線程3無法獲得鎖，線程3也直接返回數據，只有等到新開的線程2把重建數據構建完后，其他線程才能走返回正確的數據。

這種方案巧妙在于，異步的構建緩存，缺點在于在構建完緩存之前，返回的都是臟數據。

代碼實現

核心思路：相較于原來從緩存中查詢不到數據后直接查詢數據庫而言，現在的方案是 進行查詢之后，如果從緩存沒有查詢到數據，則進行互斥鎖的獲取，獲取互斥鎖后，判斷是否獲得到了鎖，如果沒有獲得到，則休眠，過一會再進行嘗試，直到獲取到鎖為止，才能進行查詢

如果獲取到了鎖的線程，再去進行查詢，查詢后將數據寫入redis，再釋放鎖，返回數據，利用互斥鎖就能保證只有一個線程去執行操作數據庫的邏輯，防止緩存擊穿

操作鎖的代碼：

核心思路就是利用redis的setnx方法來表示獲取鎖，該方法含義是redis中如果沒有這個key，則插入成功，返回1，在stringRedisTemplate中返回true， 如果有這個key則插入失敗，則返回0，在stringRedisTemplate返回false，我們可以通過true，或者是false，來表示是否有線程成功插入key，成功插入的key的線程我們認為他就是獲得到鎖的線程。

private boolean tryLock(String key) {Boolean flag = stringRedisTemplate.opsForValue().setIfAbsent(key, "1", 10, TimeUnit.SECONDS);return BooleanUtil.isTrue(flag);
}private void unlock(String key) {stringRedisTemplate.delete(key);
}

操作代碼：

 public Shop queryWithMutex(Long id)  {String key = CACHE_SHOP_KEY + id;// 1、從redis中查詢商鋪緩存String shopJson = stringRedisTemplate.opsForValue().get("key");// 2、判斷是否存在if (StrUtil.isNotBlank(shopJson)) {// 存在,直接返回return JSONUtil.toBean(shopJson, Shop.class);}//判斷命中的值是否是空值if (shopJson != null) {//返回一個錯誤信息return null;}// 4.實現緩存重構//4.1 獲取互斥鎖String lockKey = "lock:shop:" + id;Shop shop = null;try {boolean isLock = tryLock(lockKey);// 4.2 判斷否獲取成功if(!isLock){//4.3 失敗，則休眠重試Thread.sleep(50);return queryWithMutex(id);}//4.4 成功，根據id查詢數據庫shop = getById(id);// 5.不存在，返回錯誤if(shop == null){//將空值寫入redisstringRedisTemplate.opsForValue().set(key,"",CACHE_NULL_TTL,TimeUnit.MINUTES);//返回錯誤信息return null;}//6.寫入redisstringRedisTemplate.opsForValue().set(key,JSONUtil.toJsonStr(shop),CACHE_NULL_TTL,TimeUnit.MINUTES);}catch (Exception e){throw new RuntimeException(e);}finally {//7.釋放互斥鎖unlock(lockKey);}return shop;}