詳解Redis緩存擊穿以及解決方案

作者：智能運維小講堂 2018-11-12 11:12:46

安全網站安全存儲 Redis

我們正常人在登錄首頁的時候，都是根據userID來命中數據，然而黑客可以隨機生成一堆userID,然后將這些請求懟到你的服務器上，這些請求在緩存中不存在，就會穿過緩存，直接懟到數據庫上,從而造成數據庫連接異常。

什么是緩存擊穿

在談論緩存擊穿之前，我們先來回憶下從緩存中加載數據的邏輯，如下圖所示：

因此，如果黑客每次故意查詢一個在緩存內必然不存在的數據，導致每次請求都要去存儲層去查詢，這樣緩存就失去了意義。如果在大流量下數據庫可能掛掉。這就是緩存擊穿。

場景如下圖所示：

我們正常人在登錄首頁的時候，都是根據userID來***數據，然而黑客的目的是破壞你的系統，黑客可以隨機生成一堆userID,然后將這些請求懟到你的服務器上，這些請求在緩存中不存在，就會穿過緩存，直接懟到數據庫上,從而造成數據庫連接異常。

解決方案

在這里我們給出三套解決方案，大家根據項目中的實際情況，選擇使用。

講下述三種方案前，我們先回憶下redis的setnx方法。

SETNX key value

將 key 的值設為 value ，當且僅當 key 不存在。

若給定的 key 已經存在，則 SETNX 不做任何動作。

SETNX 是『SET if Not eXists』(如果不存在，則 SET)的簡寫。

可用版本：>= 1.0.0
時間復雜度： O(1)
返回值：設置成功，返回 1。設置失敗，返回 0 。

效果如下：

redis> EXISTS job # job 不存在 
(integer) 0 
redis> SETNX job "programmer" # job 設置成功 
(integer) 1 
redis> SETNX job "code-farmer" # 嘗試覆蓋 job ，失敗 
(integer) 0 
redis> GET job # 沒有被覆蓋 
"programmer"

1. 使用互斥鎖

該方法是比較普遍的做法，即，在根據key獲得的value值為空時，先鎖上，再從數據庫加載，加載完畢，釋放鎖。若其他線程發現獲取鎖失敗，則睡眠50ms后重試。

至于鎖的類型，單機環境用并發包的Lock類型就行，集群環境則使用分布式鎖( redis的setnx)。

集群環境的redis的代碼如下所示：

String get(String key) { 
String value = redis.get(key); 
if (value == null) { 
if (redis.setnx(key_mutex, "1")) { 
// 3 min timeout to avoid mutex holder crash 
redis.expire(key_mutex, 3 * 60) 
value = db.get(key); 
redis.set(key, value); 
redis.delete(key_mutex); 
} else { 
//其他線程休息50毫秒后重試 
Thread.sleep(50); 
get(key); 
} 
} 
}

優點

思路簡單
保證一致性

缺點

代碼復雜度增大
存在死鎖的風險

2. 異步構建緩存

在這種方案下，構建緩存采取異步策略，會從線程池中取線程來異步構建緩存，從而不會讓所有的請求直接懟到數據庫上。該方案redis自己維護一個timeout，當timeout小于System.currentTimeMillis()時，則進行緩存更新，否則直接返回value值。

集群環境的redis代碼如下所示：

String get(final String key) { 
V v = redis.get(key); 
String vvalue = v.getValue(); 
long timeout = v.getTimeout(); 
if (v.timeout <= System.currentTimeMillis()) { 
// 異步更新后臺異常執行 
threadPool.execute(new Runnable() { 
public void run() { 
String keyMutex = "mutex:" + key; 
if (redis.setnx(keyMutex, "1")) { 
// 3 min timeout to avoid mutex holder crash 
redis.expire(keyMutex, 3 * 60); 
String dbdbValue = db.get(key); 
redis.set(key, dbValue); 
redis.delete(keyMutex); 
} 
} 
}); 
} 
return value; 
}

優點

性價***，用戶無需等待

缺點

無法保證緩存一致性

3. 布隆過濾器

(1) 原理

布隆過濾器的巨大用處就是，能夠迅速判斷一個元素是否在一個集合中。因此他有如下三個使用場景：

網頁爬蟲對URL的去重，避免爬取相同的URL地址
反垃圾郵件，從數十億個垃圾郵件列表中判斷某郵箱是否垃圾郵箱(同理，垃圾短信)
緩存擊穿，將已存在的緩存放到布隆過濾器中，當黑客訪問不存在的緩存時迅速返回避免緩存及DB掛掉。

OK，接下來我們來談談布隆過濾器的原理。

其內部維護一個全為0的bit數組，需要說明的是，布隆過濾器有一個誤判率的概念，誤判率越低，則數組越長，所占空間越大。誤判率越高則數組越小，所占的空間越小。

假設，根據誤判率，我們生成一個10位的bit數組，以及2個hash函數((f_1,f_2))，如下圖所示(生成的數組的位數和hash函數的數量，我們不用去關心是如何生成的，有數學論文進行過專業的證明)。

假設輸入集合為((N_1,N_2)),經過計算(f_1(N_1))得到的數值得為2，(f_2(N_1))得到的數值為5，則將數組下標為2和下表為5的位置置為1，如下圖所示：

同理，經過計算(f_1(N_2))得到的數值得為3，(f_2(N_2))得到的數值為6，則將數組下標為3和下表為6的位置置為1，如下圖所示：

這個時候，我們有第三個數(N_3)，我們判斷(N_3)在不在集合((N_1,N_2))中，就進行(f_1(N_3)，f_2(N_3))的計算

若值恰巧都位于上圖的紅色位置中，我們則認為，(N_3)在集合((N_1,N_2))中
若值有一個不位于上圖的紅色位置中，我們則認為，(N_3)不在集合((N_1,N_2))中

以上就是布隆過濾器的計算原理，下面我們進行性能測試，

(2) 性能測試

代碼如下：

a. 新建一個maven工程，引入guava包

<dependencies> 
<dependency> 
<groupId>com.google.guava</groupId> 
<artifactId>guava</artifactId> 
<version>22.0</version> 
</dependency> 
</dependencies>

b. 測試一個元素是否屬于一個百萬元素集合所需耗時

package bloomfilter; 
import com.google.common.hash.BloomFilter; 
import com.google.common.hash.Funnels; 
import java.nio.charset.Charset; 
public class Test { 
private static int size = 1000000; 
private static BloomFilter<Integer> bloomFilter =BloomFilter.create(Funnels.integerFunnel(), size); 
public static void main(String[] args) { 
for (int i = 0; i < size; i++) { 
bloomFilter.put(i); 
} 
long startTime = System.nanoTime(); // 獲取開始時間 
//判斷這一百萬個數中是否包含29999這個數 
if (bloomFilter.mightContain(29999)) { 
System.out.println("***了"); 
} 
long endTime = System.nanoTime(); // 獲取結束時間 
System.out.println("程序運行時間： " + (endTime - startTime) + "納秒"); 
} 
}

輸出如下所示：

***了 
程序運行時間： 219386納秒

也就是說，判斷一個數是否屬于一個***別的集合，只要0.219ms就可以完成，性能***。

c. 誤判率的一些概念

首先，我們先不對誤判率做顯示的設置，進行一個測試，代碼如下所示：

package bloomfilter; 
import java.util.ArrayList; 
import java.util.List; 
import com.google.common.hash.BloomFilter; 
import com.google.common.hash.Funnels; 
public class Test { 
private static int size = 1000000; 
private static BloomFilter<Integer> bloomFilter =BloomFilter.create(Funnels.integerFunnel(), size); 
public static void main(String[] args) { 
for (int i = 0; i < size; i++) { 
bloomFilter.put(i); 
} 
List<Integer> list = new ArrayList<Integer>(1000); 
//故意取10000個不在過濾器里的值，看看有多少個會被認為在過濾器里 
for (int i = size + 10000; i < size + 20000; i++) { 
if (bloomFilter.mightContain(i)) { 
list.add(i); 
} 
} 
System.out.println("誤判的數量：" + list.size()); 
} 
}

輸出結果如下：

誤判對數量：330

如果上述代碼所示，我們故意取10000個不在過濾器里的值，卻還有330個被認為在過濾器里，這說明了誤判率為0.03.即，在不做任何設置的情況下，默認的誤判率為0.03。

下面上源碼來證明：

接下來我們來看一下，誤判率為0.03時，底層維護的bit數組的長度如下圖所示：

將bloomfilter的構造方法改為：

private static BloomFilter<Integer> bloomFilter = BloomFilter.create(Funnels.integerFunnel(), size,0.01);

即，此時誤判率為0.01。在這種情況下，底層維護的bit數組的長度如下圖所示

由此可見，誤判率越低，則底層維護的數組越長，占用空間越大。因此，誤判率實際取值，根據服務器所能夠承受的負載來決定，不是拍腦袋瞎想的。

(3) 實際使用

redis偽代碼如下所示：

String get(String key) { 
String value = redis.get(key); 
if (value == null) { 
if(!bloomfilter.mightContain(key)){ 
return null; 
}else{ 
value = db.get(key); 
redis.set(key, value); 
} 
} 
return value； 
}

優點

思路簡單
保證一致性
性能強

缺點

代碼復雜度增大
需要另外維護一個集合來存放緩存的Key
布隆過濾器不支持刪值操作

責任編輯：趙寧寧來源：今日頭條

緩存擊穿數據庫服務器安全