Redis的緩存穿透，緩存擊穿，緩存雪崩原因及解決方案

1、前言

在我們日常的開發中，通常都是使用資料庫來進行資料的存儲，由于一般的Web系統中通常不會存在高并發的情況，是以并沒有什麼問題。可是，一旦出現大并發量的資料請求，比如一些商品搶購的情景，或者是節假日通路量瞬間變大的時候，單一使用資料庫來儲存資料的系統會因為磁盤讀/寫速度比較慢的問題而存在嚴重的性能弊端，一瞬間成千上萬的請求到來，需要系統在極短的時間内完成成千上萬次的讀/寫操作，這個時候資料庫往往不能夠承受，極其容易造成資料庫系統癱瘓，最終導緻伺服器當機的嚴重生産事故。

為了克服上述的問題，開發人員通常會引入NoSQL技術，這是一種基于記憶體的資料庫，并且提供一定的持久化功能。Redis技術就是NoSQL技術中的集大成者，但是引入Redis又有可能出現緩存穿透，緩存擊穿，緩存雪崩等問題。本文就對這三種問題進行較深入剖析。

2、初認識緩存穿透，緩存擊穿，緩存雪崩

緩存穿透：Key對應的資料在緩存中并不存在，每次針對此Key的請求從緩存擷取不到，請求都會到資料庫，進而可能壓垮資料庫。比如，用一個不存在的使用者ID擷取使用者資訊，不論緩存還是資料庫都沒有，若黑客利用此漏洞進行攻擊可能壓垮資料庫。

緩存擊穿：Key對應的資料存在，并且屬于熱點，熱點Key在某個時間點過期的時候，此時若有大量并發請求過來，這些請求發現緩存過期一般都會從後端資料庫加載資料并回設到緩存，這個時候大并發的請求可能會瞬間把後端資料庫壓垮，如同在一個屏障上鑿開了一個洞，是以稱之為“緩存擊穿”。

緩存雪崩：當緩存伺服器重新開機或者大量緩存Key集中在某一個時間段失效，所有的查詢都落在資料庫上，造成了緩存雪崩，也會給後端資料庫帶來很大壓力。

3、緩存穿透解決方案

如果從存儲層查不到資料則不寫入緩存，這将導緻這個不存在的資料每次請求都要到存儲層去查詢，失去了緩存的意義。

有很多種方法可以有效地解決緩存穿透問題，最常見的則是采用布隆過濾器，将所有可能存在的資料哈希到一個足夠大的BitMap中，一定不存在的資料會被這個BitMap攔截掉，進而避免了對底層存儲系統的查詢壓力。

布隆過濾器（Bloom filter）簡介

Bloom filter 是由 Howard Bloom 在 1970 年提出的二進制向量資料結構，它具有很好的空間和時間效率，被用來檢測一個元素是不是集合中的一個成員。

如需要判斷一個元素是不是在一個集合中，我們通常做法是把所有元素儲存下來，然後通過比較知道它是不是在集合内，連結清單、樹都是基于這種思路，當集合内元素個數的變大，我們需要的空間和時間都線性變大，檢索速度也越來越慢。 Bloom filter 采用的是哈希函數的方法，将一個元素映射到一個 m 長度的陣列上的一個點，當這個點是 1 時，那麼這個元素在集合内，反之則不在集合内。這個方法的缺點就是當檢測的元素很多的時候可能有沖突，解決方法就是使用 k 個哈希函數對應 k 個點，如果所有點都是 1 的話，那麼元素在集合内，如果有 0 的話，元素則不在集合内。

備注：如果檢測結果為“是”，該元素不一定在集合中，但如果檢測結果為“否”，該元素一定不在集合中。

另外，也有一個更為簡單粗暴的方法，如果一個查詢傳回的資料為空（不管是資料不存在，還是系統故障），我們仍然把這個空結果進行緩存，但它的過期時間會很短，最長不超過五分鐘。

//僞代碼，秒殺蘋果手機
public Object getIPhone()
{
int cacheTime = 300;
String cacheKey = "iPhone";
String cacheValue = CacheHelper.get(cacheKey);
if (cacheValue != null)
{
return cacheValue;
}
else
{
//查詢資料庫
cacheValue = DBHelper.getIPhoneFromDB();
if (cacheValue == null)
{
//如果發現為空，設定個預設值，也緩存起來
cacheValue = "";
}
CacheHelper.add(cacheKey, cacheValue, cacheTime);
return cacheValue;
}
}

4、緩存擊穿解決方案

業界比較常用的做法是使用Redis的互斥鎖(mutex key)。簡單地來說，就是在緩存失效的時候（判斷拿出來的值為空），不是立即去查詢資料庫加載資料，而是先去set一個mutex key，當操作傳回成功時（意味着獲得了互斥鎖），再進行查庫操作并回設緩存；否則，就重試整個get緩存的方法。

SETNX 是「SET if Not eXists」的縮寫，也就是隻有不存在的時候才設定，可以利用它來實作鎖的效果。

public Object getIPhone()
{
int cacheTime = 300;
String cacheKey = "iPhone";
String cacheValue = CacheHelper.get(cacheKey);
if (cacheValue != null)
{
return cacheValue;
}
else
{
//查詢資料庫，先擷取鎖，需要分情況考慮
String lockKey = "lock";
//需要設定過期時間，防止此線程挂掉之後，其他線程也無法加鎖
Boolean lockResult = CacheHelper.setnx(lockKey,"mylocker",30)
if (lockResult)
{//情況1：加鎖成功
cacheValue = DBHelper.getIPhoneFromDB();
if (cacheValue == null)
{
//如果發現為空，設定個預設值，也緩存起來
cacheValue = "";
}
CacheHelper.add(cacheKey, cacheValue, cacheTime);
CacheHelper.delete(lockKey);
return cacheValue;
}
else
{//情況2：加鎖失敗
Thread.sleep(30);
cacheValue = CacheHelper.get(cacheKey);
return cacheValue;
}
}
}

上述代碼看似完美，但是存在問題：雖然加鎖成功了，但如果查庫的時間過長，導緻鎖失效了，最後delete鎖的時候，删掉的是其他線程加的鎖。這個時候應該是“各加各鎖，各删各鎖”，如下所示：

//查詢資料庫，先擷取鎖，需要分情況考慮
String lockKey = "lock";
String lockValue = UUID.randomUUID().toString();
//需要設定過期時間，防止此線程挂掉之後，其他線程也無法加鎖
Boolean lockResult = CacheHelper.setnx(lockKey,lockValue,30)
if (lockResult)
{//情況1：加鎖成功
cacheValue = DBHelper.getIPhoneFromDB();
if (cacheValue == null)
{
//如果發現為空，設定個預設值，也緩存起來
cacheValue = "";
}
CacheHelper.add(cacheKey, cacheValue, cacheTime);
if (lockValue.equals(CacheHelper.get(lockKey)))
{
CacheHelper.delete(lockKey);
}
return cacheValue;
}

5、緩存雪崩解決方案

緩存失效時的雪崩效應對底層系統的沖擊非常可怕，大多數系統設計者考慮用加鎖或者隊列的方式保證來保證不會有大量的線程對資料庫一次性進行讀寫，進而避免失效時大量的并發請求落到底層存儲系統上。如下所示：

public Object getIPhone()
{
int cacheTime = 300;
String cacheKey = "iPhone";
String cacheValue = CacheHelper.get(cacheKey);
if (cacheValue != null)
{
return cacheValue;
}
else
{
synchronized(Object.class)
{//加鎖之後，形成一個等待隊列
cacheValue = CacheHelper.get(cacheKey);
if (cacheValue != null)
{
return cacheValue;
}
else
{
cacheValue = DBHelper.getIPhoneFromDB();
if (cacheValue == null)
{
//如果發現為空，設定個預設值，也緩存起來
cacheValue = "";
}
CacheHelper.add(cacheKey, cacheValue, cacheTime);
return cacheValue;
}
}
}
}

加鎖排隊隻是為了減輕資料庫的壓力，并沒有提高系統吞吐量。假設在高并發下，緩存重建期間，這時過來1000個請求有999個都在阻塞的。同樣會導緻使用者等待逾時，這是個治标不治本的方法！如下所示：

還有一個簡單方案就時講緩存失效時間分散開，比如我們可以在原有的失效時間基礎上增加一個随機值，比如1-5分鐘随機時間，這樣每一個緩存的過期時間的重複率就會降低，就很難引發集體失效的事件。

還有一種被稱為“二級緩存”的解決方法，二級緩存過期時間比一級緩存的時間延長1倍，當一級緩存失效的時候，傳回二級緩存的資料，并異步啟動一個新線程去重建緩存。如下代碼所示：

public Object getIPhone()
{
int cacheTime = 300;
String cacheKey1 = "iPhone1";
String cacheKey2 = "iPhone2";
String cacheValue1 = CacheHelper.get(cacheKey1);
String cacheValue2 = CacheHelper.get(cacheKey2);
if(cacheValue1 != null)
{//一級緩存有效，則傳回一級緩存的值
return cacheValue1;
}