作者：小程故事多
一、前言
我們在用緩存的時候，不管是Redis或者Memcached，基本上會通用遇到以下三個問題：
- 緩存穿透
- 緩存并發
- 緩存失效
注：
上面三個圖會有什麼問題呢？
我們在項目中使用緩存通常都是先檢查緩存中是否存在，如果存在直接傳回緩存内容，如果不存在就直接查詢資料庫然後再緩存查詢結果傳回。這個時候如果我們查詢的某一個資料在緩存中一直不存在，就會造成每一次請求都查詢DB，這樣緩存就失去了意義，在流量大時，可能DB就挂掉了。那這種問題有什麼好辦法解決呢？
要是有人利用不存在的key頻繁進攻我們的應用，這就是漏洞。
有一個比較巧妙的作法是，可以将這個不存在的key預先設定一個值。
比如，"key" , “&&”。
在傳回這個&&值的時候，我們的應用就可以認為這是不存在的key，那我們的應用就可以決定是否繼續等待繼續通路，還是放棄掉這次操作。如果繼續等待通路，過一個時間輪詢點後，再次請求這個key，如果取到的值不再是&&，則可以認為這時候key有值了，進而避免了透傳到資料庫，進而把大量的類似請求擋在了緩存之中。
有時候如果網站并發通路高，一個緩存如果失效，可能出現多個程序同時查詢DB，同時設定緩存的情況，如果并發确實很大，這也可能造成DB壓力過大，還有緩存頻繁更新的問題。
我現在的想法是對緩存查詢加鎖，如果KEY不存在，就加鎖，然後查DB入緩存，然後解鎖；其他程序如果發現有鎖就等待，然後等解鎖後傳回資料或者進入DB查詢。
這種情況和剛才說的預先設定值問題有些類似，隻不過利用鎖的方式，會造成部分請求等待。
引起這個問題的主要原因還是高并發的時候，平時我們設定一個緩存的過期時間時，可能有一些會設定1分鐘啊，5分鐘這些，并發很高時可能會出在某一個時間同時生成了很多的緩存，并且過期時間都一樣，這個時候就可能引發一當過期時間到後，這些緩存同時失效，請求全部轉發到DB，DB可能會壓力過重。
那如何解決這些問題呢？
其中的一個簡單方案就時講緩存失效時間分散開，比如我們可以在原有的失效時間基礎上增加一個随機值，比如1-5分鐘随機，這樣每一個緩存的過期時間的重複率就會降低，就很難引發集體失效的事件。
我們讨論的第二個問題時針對同一個緩存，第三個問題時針對很多緩存。
總結來看：
- 緩存穿透：查詢一個必然不存在的資料。比如文章表，查詢一個不存在的id，每次都會通路DB，如果有人惡意破壞，很可能直接對DB造成影響。
- 緩存失效：如果緩存集中在一段時間内失效，DB的壓力凸顯。這個沒有完美解決辦法，但可以分析使用者行為，盡量讓失效時間點均勻分布。
當發生大量的緩存穿透，例如對某個失效的緩存的大并發通路就造成了緩存雪崩。
問題彙總
問題1：
如何解決DB和緩存一緻性問題？
答：當修改了資料庫後，有沒有及時修改緩存。這種問題，以前有過實踐，修改資料庫成功，而修改緩存失敗的情況，最主要就是緩存伺服器挂了。而因為網絡問題引起的沒有及時更新，可以通過重試機制來解決。而緩存伺服器挂了，請求首先自然也就無法到達，進而直接通路到資料庫。那麼我們在修改資料庫後，無法修改緩存，這時候可以将這條資料放到資料庫中，同時啟動一個異步任務定時去檢測緩存伺服器是否連接配接成功，一旦連接配接成功則從資料庫中按順序取出修改資料，依次進行緩存最新值的修改。
問題2：
問下緩存穿透那塊！例如，一個使用者查詢文章，通過ID查詢，按照之前說的，是将緩存的KEY預先設定一個值，，如果通過ID插過來，發現是預先設定的一個值，比如說是“&&”，那之後的繼續等待通路是什麼意思，這個ID什麼時候會真正被附上使用者所需要的值呢？
答：我剛說的主要是咱們常用的後面配置，前台擷取的場景。前台無法擷取相應的key，則等待，或者放棄。當在背景配置界面上配置了相關key和value之後，那麼以前的key &&也自然會被替換掉。你說的那種情況，自然也應該會有一個程序會在某一個時刻，在緩存中設定這個ID，再有新的請求到達的時候，就會擷取到最新的ID和value。
問題3：
其實用redis的話，那天看到一個不錯的例子，雙key，有一個當時生成的一個附屬key來辨別資料修改到期時間，然後快到的時候去重新加載資料，如果覺得key多可以把結束時間放到主key中，附屬key起到鎖的功能。
答：這種方案，之前我們實踐過。這種方案會産生雙份資料，而且需要同時控制附屬key與key之間的關系，操作上有一定複雜度。
問題4：
多級緩存是什麼概念呢？
答：多級緩存就像我今天之前給大家發的文章裡面提到了，将ehcache與redis做二級緩存，就像我之前寫的文章 http://www.jianshu.com/p/2cd6ad416a5a 提到過的。但同樣會存在一緻性問題，如果我們需要強一緻性的話，緩存與資料庫同步是會存在時間差的，是以我們在具體開發的過程中，一定要根據場景來具體分析，二級緩存更多的解決是，緩存穿透與程式的健壯性，當集中式緩存出現問題的時候，我們的應用能夠繼續運作。
說明：本文中提到的緩存可以了解為Redis。
二、緩存穿透與并發方案
上文中介紹了關于緩存穿透、并發的一些常用思路，但是沒有明确一些思路的使用場景，下面繼續深入探讨。相信不少朋友之前看過很多類似的文章，但是歸根結底就是二個問題：
- 如何解決穿透
- 如何解決并發
當并發較高的時候，其實我是不建議使用緩存過期這個政策的，我更希望緩存一直存在，通過背景系統來更新緩存系統中的資料達到資料的一緻性目的，有的朋友可能會質疑，如果緩存系統挂了怎麼辦，這樣資料庫更新了但是緩存沒有更新，沒有達到一緻性的狀态。
解決問題的思路是：如果緩存是因為網絡問題沒有更新成功資料，那麼建議重試幾次，如果依然沒有更新成功則認為緩存系統出錯不可用，這時候用戶端會将資料的KEY插入到消息系統中，消息系統可以過濾相同的KEY，隻需保證消息系統不存在相同的KEY，當緩存系統恢複可用的時候，依次從mq中取出KEY值然後從資料庫中讀取最新的資料更新緩存。注意：更新緩存之前，緩存中依然有舊資料，是以不會造成緩存穿透。
下圖展示了整個思路的過程：
看完上面的方案以後，又會有不少朋友提出疑問，如果我是第一次使用緩存或者緩存中暫時沒有我需要的資料，那又該如何處理呢？
解決問題的思路：在這種場景下，用戶端從緩存中根據KEY讀取資料，如果讀到了資料則流程結束，如果沒有讀到資料（可能會有多個并發都沒有讀到資料），這時候使用緩存系統中的setNX方法設定一個值（這種方法類似加個鎖），沒有設定成功的請求則sleep一段時間，設定成功的請求讀取資料庫擷取值，如果擷取到則更新緩存，流程結束，之前sleep的請求這時候喚醒後直接再從緩存中讀取資料，此時流程結束。
在看完這個流程後，我想這裡面會有一個漏洞，如果資料庫中沒有我們需要的資料該怎麼處理，如果不處理則請求會造成死循環，不斷的在緩存和資料庫中查詢，這時候我們會沿用我之前文章中的如果沒有讀到資料則往緩存中插入一個NULL字元串的思路，這樣其他請求直接就可以根據“NULL”進行處理，直到背景系統在資料庫成功插入資料後同步更新清理NULL資料和更新緩存。
流程圖如下所示：
總結：在實際工作中，我們往往将上面二個方案組合使用才能達到最佳效果，雖然第二種方案也會造成請求阻塞，但是隻是在第一次使用或者緩存暫時沒有資料的情況下才會産生，在生産中經過檢驗在TPS沒有上萬的情況下是不會造成問題的。
三、熱點緩存解決方案
1、緩存使用背景：
我們拿使用者中心的一個案例來說明：每個使用者都會首先擷取自己的使用者資訊，然後再進行其他相關的操作，有可能會有如下一些場景情況：
- 會有大量相同使用者重複通路該項目。
- 會有同一使用者頻繁通路同一子產品。
2、思路解析
- 因為使用者本身是不固定的而且使用者數量也有幾百萬尤其上千萬，我們不可能把所有的使用者資訊全部緩存起來，通過第一個場景情況可以看到一些規律，那就是有大量的相同使用者重複通路，但是究竟是哪些使用者重複通路我們也并不知道。
- 如果有一個使用者頻繁重新整理讀取項目，那麼對資料庫本身也會造成較大壓力，當然我們也會有相關的保護機制來确實惡意進攻，可以從前端控制，也可以有采黑名單等機制，這裡不在贅述。如果用緩存的話，我們又該如何控制同一使用者繁重讀取使用者資訊呢。
請看下圖：
for (int i = 0; i < times; i++) {
user = new ExternalUser();
user.setId(i+"");
user.setUpdateTime(new Date(System.currentTimeMillis()));
CacheUtil.zadd(sortKey, user.getUpdateTime().getTime(), user.getId());
CacheUtil.putAndThrowError(userKey+user.getId(), JSON.toJSONString(user));
}
Set<String> userSet = CacheUtil.zrange(sortKey, 0, -1);
System.out.println("[sortedSet] - " + JSON.toJSONString(userSet) );
if(userSet == null || userSet.size() == 0)
return;
Set<Tuple> userSetS = CacheUtil.zrangeWithScores(sortKey, 0, -1);
StringBuffer sb = new StringBuffer();
for(Tuple t:userSetS){
sb.append("{member: ").append(t.getElement()).append(", score: ").append(t.getScore()).append("}, ");
}
System.out.println("[sortedcollect] - " + sb.toString().substring(0, sb.length() - 2));
Set<String> members = new HashSet<String>();
for(String uid:userSet){
String key = userKey + uid;
members.add(uid);
ExternalUser user2 = CacheUtil.getObject(key, ExternalUser.class);
System.out.println("[user] - " + JSON.toJSONString(user2) );
}
System.out.println("[user] - " + System.currentTimeMillis());
String[] keys = new String[members.size()];
members.toArray(keys);
Long rem = CacheUtil.zrem(sortKey, keys);
System.out.println("[rem] - " + rem);
userSet = CacheUtil.zrange(sortKey, 0, -1);
System.out.println("[remove - sortedSet] - " + JSON.toJSONString(userSet));