作者簡介：曾任職于阿裡巴巴，每日優鮮等網際網路公司，任技術總監，15年電商網際網路經曆。
近些年，Redis憑借在性能、穩定性和高可擴充性上的卓越表現，基本上已經成了網際網路行業緩存中間件的标配，甚至很多傳統行業也在使用Redis。那麼我們在使用Redis等緩存中間件時，要注意哪些問題呢？本文咱們就來聊聊，我們使用緩存中間件過程中曾經遇到的坑！
緩存穿透
先看一個常見的緩存使用方式。請求來了，先查緩存，緩存有值就直接傳回；緩存沒值，查資料庫，然後把資料庫的值存到緩存，再傳回。
假如緩存沒查到某個值，查資料庫也沒這個值，也就是說要查的值根本不存在，這樣就會導緻每次對這個值的查詢請求都會穿透到資料庫。這就是所謂的“緩存穿透”。
如何避免緩存穿透？
如果從資料庫中沒查到值，可以在緩存中記錄一個空值，來避免“緩存穿透”。并且要給這個空值設定一個較短的過期時間。
比如說，我們經常會把使用者資訊緩存到Redis。如果調用方傳了一個不存在的UserID，在緩存中就查不到這個使用者資訊，然後去DB也查不到。這樣就會導緻，每次根據這個UserID查使用者資訊，都會穿透到資料庫，給資料庫造成了壓力。為了避免緩存穿透，當資料庫查不到時，我們可以在緩存中記錄一條空資料，比如userID做為key，空json做為值，如果程式獲得這個空json，就按使用者不存在處理。再給這個key設定一個很短的過期時間，比如30秒。
緩存雪崩
我們經常會遇到需要初始化緩存的情況。比如說使用者系統重構，表結構發生了變化，緩存資訊也要變，上線前需要初始化緩存，将使用者資訊批量存入緩存。假如我們給這些使用者資訊設定相同的過期時間，到過期時間點所有使用者資訊的緩存記錄就會同時集中失效，導緻大量請求瞬間打到資料庫，資料庫很可能會被搞挂。這種緩存集中失效，導緻大量請求同時穿透到資料庫的情況，就是所謂的“雪崩效應”。
是以，當我們向緩存初始化資料時，要保證每個緩存記錄過期時間的離散性。可以采用一個較大的固定值加上一個較小的随機值。比如過期時間可以是：10小時 + 0到3600秒的随機值。
緩存并發
當系統并發很高，緩存資料尤其是熱點資料過期後，可能會出現多個請求同時通路資料庫并設定緩存的情況，不但給資料庫帶來壓力，而且會有緩存頻繁更新的問題。
我們可以通過加鎖來避免緩存并發問題。如果從緩存查不到資料，對查詢資料加分布式鎖，然後查資料庫并把資料庫查詢結果放入緩存。其他線程等待鎖釋放後，直接從緩存取值。
比如，電商系統會緩存商品SKU價格，一些熱點商品的并發通路會非常高。當緩存過期失效後，通路請求從緩存查不到記錄，此時可以用商品SKU ID為Key加分布式鎖，然後從資料庫查詢價格并把價格放入緩存，最後解鎖。解鎖後其他請求就可以從緩存直接取值了。進而避免了資料庫的壓力。
分布式鎖
以我們之前做過的5人拼團為例。如果有使用者參加團購，我們需要先校驗參團人數是否達到了上限5人。如果沒達到5人，使用者才可以參團。僞代碼如下：
//根據拼團ID擷取目前參團成員數量 int numOfMembers = pinTuanService.getNumOfMembersById(pinTuanID); if(numOfMembers < 5) { pinTuanService.pintuan();//執行，加入拼團，生單等邏輯 } 高并發場景下，上面的代碼會有很嚴重的問題。如果某個團目前的參團人數是4，這時有兩個使用者同時參團，使用者A和使用者B的請求同時進入上面的代碼塊，A和B的請求同時執行到第2行代碼，擷取的numOfMembers都是4，表達式 numOfMembers < 5 成立，是以兩個使用者都能執行到第4行代碼，就是說A使用者和B使用者都能成功參加拼團。于是，參團人數就超過了5人的上限。是以我們就需要加鎖來避免這個問題。synchronized行嗎？不行。因為我們的服務是多節點部署的，是以要加分布式鎖。代碼如下：
boolean aquired = distributedLock.aquireLock(pinTuanID, 3000); if(aquired == true) { try{ //根據拼團ID擷取目前參團成員數量 int numOfMembers = pinTuanService.getNumOfMembersById(pinTuanID); if(numOfMembers < 5) { pinTuanService.pintuan();//執行，加入拼團，生單等邏輯 } } finally { distributedLock.releaseLock(pinTuanID); } } 這樣就好多啦！接下來我們看看基于Redis分布式鎖的實作，以及特别要注意的問題。一般我們會基于setnx實作Redis分布式鎖。setnx指令可以檢查key是否存在，如果key不存在，就在Redis中建立一個鍵值對（操作成功），如果key已經存在就放棄執行（操作失敗）。
先看一段基于Springboot實作的加鎖和釋放鎖的代碼：
@Component public class DistributedLock { @Autowired private StringRedisTemplate redisTemplate; /** * 加鎖 * lockKey，redis的key * expireTime，過期時間，機關是毫秒 * 注：setIfAbsent方法就使用了redis的setnx */ public boolean aquireLock(String lockKey, long expireTime) { long waitTime = 0; boolean success = redisTemplate.opsForValue().setIfAbsent(lockKey, "distributedLock", expireTime, TimeUnit.MILLISECONDS); if(success == true){ return success; } else { //如果加鎖失敗，循環重試加鎖 while(success != true && waitTime < 5000L ) { success = redisTemplate.opsForValue().setIfAbsent(lockKey, "distributedLock", expireTime, TimeUnit.MILLISECONDS); sleep 100毫秒; waitTime += 100L; } } return success; } /** * 釋放鎖 * lockKey，redis的key */ public void releaseLock(String lockKey) { redisTemplate.delete(lockKey); } } 上面的代碼。乍一看，好像沒什麼問題！加鎖失敗有循環重試加鎖，過期時間設定了，而且也保證了建立Key-Value鍵值對和設定過期時間的原子性，這樣當程式沒有正常釋放鎖時，也能保證過期後鎖自動釋放（注意：redis較老的版本不支援 setnx 和設定過期時間的原子操作，不過可以利用Lua腳本來保證原子性）。
我們再仔細思考一下，一般場景我們會對Key設定一個很短的過期時間，當一次操作因為網絡等原因耗費了較長時間，操作還沒完成key就過期失效了。這樣會産生什麼問題呢？我們還是以拼團為例加以說明，先看看下面這張圖：
如上圖，使用者A和使用者B同時參加同一團，團ID為 001，我們以團ID作為分布式鎖的Key，"distributedLock" 作為固定的Value，過期時間是5秒。A先擷取分布式鎖，但是由于網絡等原因A的拼團操作在5秒内沒完成，這時Key過期并從Redis清除掉，A的分布式鎖失效。此時使用者B拿到分布式鎖，Key也同樣是團ID 001。在使用者B的拼團邏輯執行完之前，使用者A的邏輯先執行完了，緊接着A就把鎖給釋放了。不過A的鎖早已經過期失效了，B持有鎖的Key和A又完全一樣，是以此時A釋放的其實是B的鎖。這樣一來整個拼團還是有可能會超員。怎麼解決呢？
我們可以把分布式鎖的Value設成可以區分的值，比如拼團的場景Value可以設定為userID，在釋放鎖的時候根據key和value來判斷目前的鎖是不是自己的，隻有Redis中userID和自己的userID相同才釋放鎖。
改進後的代碼如下：
@Component public class DistributedLock { @Autowired private StringRedisTemplate redisTemplate; /** * 加鎖 * lockKey，redis的key * expireTime，過期時間，機關是毫秒 * 注：setIfAbsent方法就使用了redis的setnx */ public boolean aquireLock(String lockKey, String userID, long expireTime) { long waitTime = 0; boolean success = redisTemplate.opsForValue().setIfAbsent(lockKey, userID, expireTime, TimeUnit.MILLISECONDS); if(success == true){ return success; } else { //如果加鎖失敗，循環重試加鎖 while(success != true && waitTime < 5000L ) { success = redisTemplate.opsForValue().setIfAbsent(lockKey, userID, expireTime, TimeUnit.MILLISECONDS); sleep 100毫秒; waitTime += 100L; } } return success; } /** * 釋放鎖 * lockKey，redis的key */ public void releaseLock(String lockKey, String userID) { String userIDFromRedis = redisTemplate.get(lockKey); if( userID.equals(userIDFromRedis) ) { redisTemplate.delete(lockKey); } } } 還有一種場景需要考慮。當Redis master發生故障，主備切換時往往會造成資料丢失，包括分布式鎖的Key-Value 也可能丢失。這樣就會導緻操作還沒執行完，鎖就被其他請求拿到了。Redis官方提供了Redlock算法，以及相應的開源實作 Redisson。用到分布式鎖的場景，大家可以直接使用 Redisson，非常友善。如果系統對可靠性要求很高，如需用到分布式鎖，建議使用 Zookeeper，etcd 等。
OK，就分享到這。
