1 導讀

Redis 是當前最流行的 NoSQL 數據庫。Redis 主要用來做緩存使用，在提高數據查詢效率、保護數據庫等方面起到了關鍵性的作用，很大程度上提高系統的性能。當然在使用過程中，也會出現一些異常情景，導致 Redis 失去緩存作用。

2 異常類型

異常主要有 緩存雪崩 緩存穿透 緩存擊穿。

2.1 緩存雪崩

2.1.1 現象

緩存雪崩是指大量請求在緩存中沒有查到數據，直接訪問數據庫，導致數據庫壓力增大，最終導致數據庫崩潰，從而波及整個系統不可用，好像雪崩一樣。

2.1.2 異常原因

緩存服務不可用。

緩存服務可用，但是大量 KEY 同時失效。

2.1.3 解決方案

1. 緩存服務不可用

redis 的部署方式主要有單機、主從、哨兵和 cluster 模式。

單機

只有一臺機器，所有數據都存在這臺機器上，當機器出現異常時，redis 將失效，可能會導致 redis 緩存雪崩。

主從

主從其實就是一臺機器做主，一個或多個機器做從，從節點從主節點復制數據，可以實現讀寫分離，主節點做寫，從節點做讀。
優點：當某個從節點異常時，不影響使用。
缺點：當主節點異常時，服務將不可用。

哨兵

哨兵模式也是一種主從，只不過增加了哨兵的功能，用于監控主節點的狀態，當主節點宕機之后會進行投票在從節點中重新選出主節點。
優點：高可用，當主節點異常時，自動在從節點當中選擇一個主節點。
缺點：只有一個主節點，當數據比較多時，主節點壓力會很大。

cluster 模式

集群采用了多主多從，按照一定的規則進行分片，將數據分別存儲，一定程度上解決了哨兵模式下單機存儲有限的問題。
優點：高可用，配置了多主多從，可以使數據分區，去中心化，減小了單臺機子的負擔.
缺點：機器資源使用比較多，配置復雜。

小結

從高可用得角度考慮，使用哨兵模式和 cluster 模式可以防止因為 redis 不可用導致的緩存雪崩問題。

2. 大量 KEY 同時失效

可以通過設置永不失效、設置不同失效時間、使用二級緩存和定時更新緩存失效時間

設置永不失效

如果所有的 key 都設置不失效，不就不會出現因為 KEY 失效導致的緩存雪崩問題了。redis 設置 key 永遠有效的命令如下：
PERSIST key
缺點：會導致 redis 的空間資源需求變大。

設置隨機失效時間

如果 key 的失效時間不相同，就不會在同一時刻失效，這樣就不會出現大量訪問數據庫的情況。
redis 設置 key 有效時間命令如下：
Expire key
示例代碼如下，通過 RedisClient 實現

/**
* 隨機設置小于30分鐘的失效時間
* @param redisKey
* @param value
*/
private void setRandomTimeForReidsKey(String redisKey,String value){
//隨機函數
Random rand = new Random();
//隨機獲取30分鐘內（30*60）的隨機數
int times = rand.nextInt(1800);
//設置緩存時間（緩存的key,緩存的值，失效時間：單位秒）
redisClient.setNxEx(redisKey,value,times);
}

使用二級緩存

二級緩存是使用兩組緩存，1 級緩存和 2 級緩存，同一個 Key 在兩組緩存里都保存，但是他們的失效時間不同，這樣 1 級緩存沒有查到數據時，可以在二級緩存里查詢，不會直接訪問數據庫。
示例代碼如下：

public static void main(String[] args) {
CacheTest test = new CacheTest();
//從1級緩存中獲取數據
String value = test.queryByOneCacheKey("key");
//如果1級緩存中沒有數據，再二級緩存中查找
if(StringUtils.isBlank(value)){
value = test.queryBySecondCacheKey("key");
//如果二級緩存中沒有，從數據庫中查找
if(StringUtils.isBlank(value)){
value =test.getFromDb();
//如果數據庫中也沒有，就返回空
if(StringUtils.isBlank(value)){
System.out.println("數據不存在!");
}else{
//二級緩存中保存數據
test.secondCacheSave("key",value);
//一級緩存中保存數據
test.oneCacheSave("key",value);
System.out.println("數據庫中返回數據!");
}
}else{
//一級緩存中保存數據
test.oneCacheSave("key",value);
System.out.println("二級緩存中返回數據!");
}
}else {
System.out.println("一級緩存中返回數據!");
}
}

異步更新緩存時間

每次訪問緩存時，啟動一個線程或者建立一個異步任務來，更新緩存時間。

示例代碼如下：

public class CacheRunnable implements Runnable {

private ClusterRedisClientAdapter redisClient;
/**
* 要更新的key
*/
public String key;

public CacheRunnable(String key){
this.key =key;
}

@Override
public void run() {
//更細緩存時間
redisClient.expire(this.getKey(),1800);
}

public String getKey() {
return key;
}

public void setKey(String key) {
this.key = key;
}
}
public static void main(String[] args) {
CacheTest test = new CacheTest();
//從緩存中獲取數據
String value = test.getFromCache("key");
if(StringUtils.isBlank(value)){
//從數據庫中獲取數據
value = test.getFromDb("key");
//將數據放在緩存中
test.oneCacheSave("key",value);
//返回數據
System.out.println("返回數據");
}else{
//異步任務更新緩存
CacheRunnable runnable = new CacheRunnable("key");
runnable.run();
//返回數據
System.out.println("返回數據");
}
}

3. 小結

上面從服務不可用和 key 大面積失效兩個方面，列舉了幾種解決方案，上面的代碼只是提供一些思路，具體實施還要考慮到現實情況。當然也有其他的解決方案，我這里舉例是比較常用的。畢竟現實情況，千變萬化，沒有最好的方案，只有最適用的方案。

2.2 緩存穿透

2.2.1 現象

緩存穿透是指當用戶在查詢一條數據的時候，而此時數據庫和緩存卻沒有關于這條數據的任何記錄，而這條數據在緩存中沒找到就會向數據庫請求獲取數據。用戶拿不到數據時，就會一直發請求，查詢數據庫，這樣會對數據庫的訪問造成很大的壓力。

2.2.2 異常原因

非法調用

2.2.3 解決方案

1. 非法調用

可以通過緩存空值或過濾器來解決非法調用引起的緩存穿透問題。

緩存空值

當緩存和數據庫中都沒有值時，可以在緩存中存放一個空值，這樣就可以減少重復查詢空值引起的系統壓力增大，從而優化了緩存穿透問題。
示例代碼如下：

private String queryMessager(String key){
//從緩存中獲取數據
String message = getFromCache(key);
//如果緩存中沒有 從數據庫中查找
if(StringUtils.isBlank(message)){
message = getFromDb(key);
//如果數據庫中也沒有數據 就設置短時間的緩存
if(StringUtils.isBlank(message)){
//設置緩存時間（緩存的key,緩存的值，失效時間：單位秒）
redisClient.setNxEx(key,null,60);
}else{
redisClient.setNxEx(key,message,1800);
}
}
return message;
}

缺點：大量的空緩存導致資源的浪費，也有可能導致緩存和數據庫中的數據不一致。

布隆過濾器

布隆過濾器由布隆在 1970 年提出。它實際上是一個很長的二進制向量和一系列隨機映射函數。布隆過濾器可以用于檢索一個元素是否在一個集合中。是以空間換時間的算法。

布隆過濾器的實現原理是一個超大的位數組和幾個哈希函數。

假設哈希函數的個數為 3。首先將位數組進行初始化，初始化狀態的維數組的每個位都設置位 0。如果一次數據請求的結果為空，就將 key 依次通過 3 個哈希函數進行映射，每次映射都會產生一個哈希值，這個值對應位數組上面的一個點，然后將位數組對應的位置標記為 1。

當數據請求再次發過來時，用同樣的方法將 key 通過哈希映射到位數組上的 3 個點。如果 3 個點中任意一個點不為 1，則可以判斷 key 不為空。反之，如果 3 個點都為 1，則該 KEY 一定為空。

缺點：

可能出現誤判，例如 A 經過哈希函數 存到 1、3 和 5 位置。B 經過哈希函數存到 3、5 和 7 位置。C 經過哈希函數得到位置 3、5 和 7 位置。由于 3、5 和 7 都有值，導致判斷 A 也在數組中。這種情況隨著數據的增多，幾率也變大。
布隆過濾器沒法刪除數據。

布隆過濾器增強版

增強版是將布隆過濾器的 bitmap 更換成數組，當數組某位置被映射一次時就 + 1, 當刪除時就 - 1, 這樣就避免了普通布隆過濾器刪除數據后需要重新計算其余數據包 Hash 的問題，但是依舊沒法避免誤判。

布谷鳥過濾器

但是如果這兩個位置都滿了，它就不得不「鳩占鵲巢」，隨機踢走一個，然后自己霸占了這個位置。不同于布谷鳥的是，布谷鳥哈希算法會幫這些受害者（被擠走的蛋）尋找其它的窩。因為每一個元素都可以放在兩個位置，只要任意一個有空位置，就可以塞進去。所以這個傷心的被擠走的蛋會看看自己的另一個位置有沒有空，如果空了，自己挪過去也就皆大歡喜了。但是如果這個位置也被別人占了呢？好，那么它會再來一次「鳩占鵲巢」，將受害者的角色轉嫁給別人。然后這個新的受害者還會重復這個過程直到所有的蛋都找到了自己的巢為止。

缺點：

如果數組太擁擠了，連續踢來踢去幾百次還沒有停下來，這時候會嚴重影響插入效率。這時候布谷鳥哈希會設置一個閾值，當連續占巢行為超出了某個閾值，就認為這個數組已經幾乎滿了。這時候就需要對它進行擴容，重新放置所有元素。

2. 小結

以上方法雖然都有缺點，但是可以有效的防止因為大量空數據查詢導致的緩存穿透問題，除了系統上的優化，還要加強對系統的監控，發下異常調用時，及時加入黑名單。降低異常調用對系統的影響。

2.3 緩存擊穿

2.3.1 現象

key 中對應數據存在，當 key 中對應的數據在緩存中過期，而此時又有大量請求訪問該數據，緩存中過期了，請求會直接訪問數據庫并回設到緩存中，高并發訪問數據庫會導致數據庫崩潰。redis 的高 QPS 特性，可以很好的解決查數據庫很慢的問題。

但是如果我們系統的并發很高，在某個時間節點，突然緩存失效，這時候有大量的請求打過來，那么由于 redis 沒有緩存數據，這時候我們的請求會全部去查一遍數據庫，這時候我們的數據庫服務會面臨非常大的風險，要么連接被占滿，要么其他業務不可用，這種情況就是 redis 的緩存擊穿。

2.3.2 異常原因

熱點 KEY 失效的同時，大量相同 KEY 請求同時訪問。

2.3.3 解決方案

1. 熱點 key 失效

設置永不失效

如果所有的 key 都設置不失效，不就不會出現因為 KEY 失效導致的緩存雪崩問題了。redis 設置 key 永遠有效的命令如下：
PERSIST key
缺點：會導致 redis 的空間資源需求變大。

設置隨機失效時間

如果 key 的失效時間不相同，就不會在同一時刻失效，這樣就不會出現大量訪問數據庫的情況。
redis 設置 key 有效時間命令如下：

Expire key

示例代碼如下，通過 RedisClient 實現

/**
* 隨機設置小于30分鐘的失效時間
* @param redisKey
* @param value
*/
private void setRandomTimeForReidsKey(String redisKey,String value){
//隨機函數
Random rand = new Random();
//隨機獲取30分鐘內（30*60）的隨機數
int times = rand.nextInt(1800);
//設置緩存時間（緩存的key,緩存的值，失效時間：單位秒）
redisClient.setNxEx(redisKey,value,times);
}

使用二級緩存

二級緩存是使用兩組緩存，1 級緩存和 2 級緩存，同一個 Key 在兩組緩存里都保存，但是他們的失效時間不同，這樣 1 級緩存沒有查到數據時，可以在二級緩存里查詢，不會直接訪問數據庫。

示例代碼如下：

public static void main(String[] args) {
CacheTest test = new CacheTest();
//從1級緩存中獲取數據
String value = test.queryByOneCacheKey("key");
//如果1級緩存中沒有數據，再二級緩存中查找
if(StringUtils.isBlank(value)){
value = test.queryBySecondCacheKey("key");
//如果二級緩存中沒有，從數據庫中查找
if(StringUtils.isBlank(value)){
value =test.getFromDb();
//如果數據庫中也沒有，就返回空
if(StringUtils.isBlank(value)){
System.out.println("數據不存在!");
}else{
//二級緩存中保存數據
test.secondCacheSave("key",value);
//一級緩存中保存數據
test.oneCacheSave("key",value);
System.out.println("數據庫中返回數據!");
}
}else{
//一級緩存中保存數據
test.oneCacheSave("key",value);
System.out.println("二級緩存中返回數據!");
}
}else {
System.out.println("一級緩存中返回數據!");
}
}

異步更新緩存時間

每次訪問緩存時，啟動一個線程或者建立一個異步任務來，更新緩存時間。

示例代碼如下：

public class CacheRunnable implements Runnable {

private ClusterRedisClientAdapter redisClient;
/**
* 要更新的key
*/
public String key;

public CacheRunnable(String key){
this.key =key;
}

@Override
public void run() {
//更細緩存時間
redisClient.expire(this.getKey(),1800);
}

public String getKey() {
return key;
}

public void setKey(String key) {
this.key = key;
}
}
public static void main(String[] args) {
CacheTest test = new CacheTest();
//從緩存中獲取數據
String value = test.getFromCache("key");
if(StringUtils.isBlank(value)){
//從數據庫中獲取數據
value = test.getFromDb("key");
//將數據放在緩存中
test.oneCacheSave("key",value);
//返回數據
System.out.println("返回數據");

}else{
//異步任務更新緩存
CacheRunnable runnable = new CacheRunnable("key");
runnable.run();
//返回數據
System.out.println("返回數據");
}
}

分布式鎖

使用分布式鎖，同一時間只有 1 個請求可以訪問到數據庫，其他請求等待一段時間后，重復調用。

示例代碼如下：

/**
* 根據key獲取數據
* @param key
* @return
* @throws InterruptedException
*/
public String queryForMessage(String key) throws InterruptedException {
//初始化返回結果
String result = StringUtils.EMPTY;
//從緩存中獲取數據
result = queryByOneCacheKey(key);
//如果緩存中有數據，直接返回
if(StringUtils.isNotBlank(result)){
return result;
}else{
//獲取分布式鎖
if(lockByBusiness(key)){
//從數據庫中獲取數據
result = getFromDb(key);
//如果數據庫中有數據，就加在緩存中
if(StringUtils.isNotBlank(result)){
oneCacheSave(key,result);
}
}else {
//如果沒有獲取到分布式鎖，睡眠一下，再接著查詢數據
Thread.sleep(500);
return queryForMessage(key);
}
}
return result;
}

2. 小結

除了以上解決方法，還可以預先設置熱門數據，通過一些監控方法，及時收集熱點數據，將數據預先保存在緩存中。

3 總結

Redis 緩存在互聯網中至關重要，可以很大的提升系統效率。 本文介紹的緩存異常以及解決思路有可能不夠全面，但也提供相應的解決思路和代碼大體實現，希望可以為大家提供一些遇到緩存問題時的解決思路。如果有不足的地方，也請幫忙指出，大家共同進步。

審核編輯：劉清