一、前言

我們在用緩存的時候，不管是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做二級緩存，就像我之前寫的文章提到過的。但同樣會存在一致性問題，如果我們需要強一致性的話，緩存與數據庫同步是會存在時間差的，所以我們在具體開發的過程中，一定要根據場景來具體分析，二級緩存更多的解決是，緩存穿透與程序的健壯性，當集中式緩存出現問題的時候，我們的應用能夠繼續運行。

說明：本文中提到的緩存可以理解為Redis。

二、緩存穿透與并發方案

上文中介紹了關于緩存穿透、并發的一些常用思路，但是沒有明確一些思路的使用場景，下面繼續深入探討。相信不少朋友之前看過很多類似的文章，但是歸根結底就是二個問題：

• 如何解決穿透
• 如何解決并發

當并發較高的時候，其實我是不建議使用緩存過期這個策略的，我更希望緩存一直存在，通過后臺系統來更新緩存系統中的數據達到數據的一致性目的，有的朋友可能會質疑，如果緩存系統掛了怎么辦，這樣數據庫更新了但是緩存沒有更新，沒有達到一致性的狀態。

解決問題的思路是：

如果緩存是因為網絡問題沒有更新成功數據，那么建議重試幾次，如果依然沒有更新成功則認為緩存系統出錯不可用，這時候客戶端會將數據的KEY插入到消息系統中，消息系統可以過濾相同的KEY，只需保證消息系統不存在相同的KEY，當緩存系統恢復可用的時候，依次從mq中取出KEY值然后從數據庫中讀取最新的數據更新緩存。

注意：更新緩存之前，緩存中依然有舊數據，所以不會造成緩存穿透。

下圖展示了整個思路的過程： 

看完上面的方案以后，又會有不少朋友提出疑問，如果我是第一次使用緩存或者緩存中暫時沒有我需要的數據，那又該如何處理呢?

解決問題的思路：

在這種場景下，客戶端從緩存中根據KEY讀取數據，如果讀到了數據則流程結束，如果沒有讀到數據(可能會有多個并發都沒有讀到數據)，這時候使用緩存系統中的setNX方法設置一個值(這種方法類似加個鎖)，沒有設置成功的請求則sleep一段時間，設置成功的請求讀取數據庫獲取值，如果獲取到則更新緩存，流程結束，之前sleep的請求這時候喚醒后直接再從緩存中讀取數據，此時流程結束。

在看完這個流程后，我想這里面會有一個漏洞，如果數據庫中沒有我們需要的數據該怎么處理，如果不處理則請求會造成死循環，不斷的在緩存和數據庫中查詢，這時候我們會沿用我之前文章中的如果沒有讀到數據則往緩存中插入一個NULL字符串的思路，這樣其他請求直接就可以根據“NULL”進行處理，直到后臺系統在數據庫成功插入數據后同步更新清理NULL數據和更新緩存。

流程圖如下所示： 

總結：

在實際工作中，我們往往將上面二個方案組合使用才能達到最佳效果，雖然第二種方案也會造成請求阻塞，但是只是在第一次使用或者緩存暫時沒有數據的情況下才會產生，在生產中經過檢驗在TPS沒有上萬的情況下是不會造成問題的。

三、熱點緩存解決方案

1、緩存使用背景：

我們拿用戶中心的一個案例來說明：

每個用戶都會首先獲取自己的用戶信息，然后再進行其他相關的操作，有可能會有如下一些場景情況：

• 會有大量相同用戶重復訪問該項目。
• 會有同一用戶頻繁訪問同一模塊。

2、思路解析

因為用戶本身是不固定的而且用戶數量也有幾百萬尤其上千萬，我們不可能把所有的用戶信息全部緩存起來，通過第一個場景情況可以看到一些規律，那就是有大量的相同用戶重復訪問，但是究竟是哪些用戶重復訪問我們也并不知道。

如果有一個用戶頻繁刷新讀取項目，那么對數據庫本身也會造成較大壓力，當然我們也會有相關的保護機制來確實惡意攻擊，可以從前端控制，也可以有采黑名單等機制，這里不在贅述。如果用緩存的話，我們又該如何控制同一用戶繁重讀取用戶信息呢。

請看下圖： 

我們會通過緩存系統做一個排序隊列，比如1000個用戶，系統會根據用戶的訪問時間更新用戶信息的時間，越是最近訪問的用戶排名越排前，系統會定期過濾掉排名最后的200個用戶，然后再從數據庫中隨機取出200個用戶加入隊列，這樣請求每次到達的時候，會先從隊列中獲取用戶信息，如果命中則根據userId，再從另一個緩存數據結構中讀取用戶信息，如果沒有命中則說明該用戶請求頻率不高。JAVA偽代碼如下所示：

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));