【分佈式】緩存穿透、緩存雪崩，緩存擊穿解決方案

1、什麼樣的數據適合緩存

2、緩存穿透

緩存穿透是指查詢一個必定不存在的數據，因爲緩存是不命中時須要從數據庫查詢，查不到數據則不寫入緩存，這將致使這個不存在的數據每次請求都要到數據庫去查詢，形成緩存穿透。在流量大時，可能DB就掛掉了，要是有人利用不存在的key頻繁攻擊咱們的應用，這就是漏洞。

 解決方案：

1）有不少種方法能夠有效地解決緩存穿透問題，最多見的則是採用布隆過濾器，將全部可能存在的數據哈希到一個足夠大的bitmap中，一個必定不存在的數據會被這個bitmap攔截掉，從而避免了對底層數據庫的查詢壓力。

2）另外也有一個更爲簡單粗暴的方法，若是一個查詢返回的數據爲空（不論是數據不存在，仍是系統故障），仍然把這個空結果進行緩存，但它的過時時間會很短，最長不超過五分鐘。

 

3、緩存雪崩：

緩存雪崩是指在設置緩存時採用了相同的過時時間，致使緩存在某一時刻同時失效，致使全部的查詢都落在數據庫上，形成了緩存雪崩。

解決方案：

1）在緩存失效後，經過加鎖或者隊列來控制讀數據庫寫緩存的線程數量。好比對某個key只容許一個線程查詢數據和寫緩存，其餘線程等待。

2）能夠經過緩存reload機制，預先去更新緩存，在即將發生大併發訪問前手動觸發加載緩存。

3）不一樣的key，設置不一樣的過時時間，讓緩存失效的時間點儘可能均勻。

4）作二級緩存，或者雙緩存策略。A1爲原始緩存，A2爲拷貝緩存，A1失效時，能夠訪問A2，A1緩存失效時間設置爲短時間，A2設置爲長期。

4、緩存擊穿

對於一些設置了過時時間的key，若是這些key可能會在某些時間點被超高併發地訪問，是一種很是「熱點」的數據。這個時候，須要考慮一個問題：緩存被「擊穿」的問題，這個和緩存雪崩的區別在於這裏針對某一key緩存，前者則是不少key。 
緩存在某個時間點過時的時候，剛好在這個時間點對這個Key有大量的併發請求過來，這些請求發現緩存過時通常都會從後端DB加載數據並回設到緩存，這個時候大併發的請求可能會瞬間把後端DB壓垮。

解決方案：

1）後臺刷新

後臺定義一個job(定時任務)專門主動更新緩存數據.好比,一個緩存中的數據過時時間是30分鐘,那麼job每隔29分鐘定時刷新數據(將從數據庫中查到的數據更新到緩存中).

注：這種方案比較容易理解，但會增長系統複雜度。比較適合那些 key 相對固定,cache 粒度較大的業務，key 比較分散的則不太適合，實現起來也比較複雜。

2）檢查更新

將緩存key的過時時間(絕對時間)一塊兒保存到緩存中(能夠拼接,能夠添加新字段,能夠採用單獨的key保存..無論用什麼方式,只要二者創建好關聯關係就行).在每次執行get操做後,都將get出來的緩存過時時間與當前系統時間作一個對比,若是緩存過時時間-當前系統時間<=1分鐘(自定義的一個值),則主動更新緩存.這樣就能保證緩存中的數據始終是最新的(和方案一同樣,讓數據不過時.)

注：這種方案在特殊狀況下也會有問題。假設緩存過時時間是12:00，而 11:59 到 12:00這 1 分鐘時間裏剛好沒有 get 請求過來，又剛好請求都在 11:30 分的時 候高併發過來，那就悲劇了。這種狀況比較極端，但並非沒有可能。由於「高 併發」也多是階段性在某個時間點爆發。

3）分級緩存

採用 L1 (一級緩存)和 L2(二級緩存) 緩存方式，L1 緩存失效時間短，L2 緩存失效時間長。 請求優先從 L1 緩存獲取數據，若是 L1緩存未命中則加鎖，只有 1 個線程獲取到鎖,這個線程再從數據庫中讀取數據並將數據再更新到到 L1 緩存和 L2 緩存中，而其餘線程依舊從 L2 緩存獲取數據並返回。

注：這種方式，主要是經過避免緩存同時失效並結合鎖機制實現。因此，當數據更 新時，只能淘汰 L1 緩存，不能同時將 L1 和 L2 中的緩存同時淘汰。L2 緩存中 可能會存在髒數據，須要業務可以容忍這種短期的不一致。並且，這種方案 可能會形成額外的緩存空間浪費。

4）加鎖

方法1：

// 方法1:
    public synchronized List<String> getData01() {
        List<String> result = new ArrayList<String>();
        // 從緩存讀取數據
        result = getDataFromCache();
        if (result.isEmpty()) {
            // 從數據庫查詢數據
            result = getDataFromDB();
            // 將查詢到的數據寫入緩存
            setDataToCache(result);
        }
        return result;
    }　　

 

注：這種方式確實可以防止緩存失效時高併發到數據庫,可是緩存沒有失效的時候,在從緩存中拿數據時須要排隊取鎖,這必然會大大的下降了系統的吞吐量.

方法2：

// 方法2:
    static Object lock = new Object();
 
    public List<String> getData02() {
        List<String> result = new ArrayList<String>();
        // 從緩存讀取數據
        result = getDataFromCache();
        if (result.isEmpty()) {
            synchronized (lock) {
                // 從數據庫查詢數據
                result = getDataFromDB();
                // 將查詢到的數據寫入緩存
                setDataToCache(result);
            }
        }
        return result;
    }　

 

注：這個方法在緩存命中的時候,系統的吞吐量不會受影響,可是當緩存失效時,請求仍是會打到數據庫,只不過不是高併發而是阻塞而已.可是,這樣會形成用戶體驗不佳,而且還給數據庫帶來額外壓力.

方法3：

//方法3
    public List<String> getData03() {
        List<String> result = new ArrayList<String>();
        // 從緩存讀取數據
        result = getDataFromCache();
        if (result.isEmpty()) {
            synchronized (lock) {
            //雙重判斷,第二個以及以後的請求沒必要去找數據庫,直接命中緩存
                // 查詢緩存
                result = getDataFromCache();
                if (result.isEmpty()) {
                    // 從數據庫查詢數據
                    result = getDataFromDB();
                    // 將查詢到的數據寫入緩存
                    setDataToCache(result);
                }
            }
        }
        return result;
    }

 

注：雙重判斷雖然可以阻止高併發請求打到數據庫,可是第二個以及以後的請求在命中緩存時,仍是排隊進行的.好比,當30個請求一塊兒併發過來,在雙重判斷時,第一個請求去數據庫查詢並更新緩存數據,剩下的29個請求則是依次排隊取緩存中取數據.請求排在後面的用戶的體驗會不爽.

方法4：

static Lock reenLock = new ReentrantLock();
 
    public List<String> getData04() throws InterruptedException {
        List<String> result = new ArrayList<String>();
        // 從緩存讀取數據
        result = getDataFromCache();
        if (result.isEmpty()) {
            if (reenLock.tryLock()) {
                try {
                    System.out.println("我拿到鎖了,從DB獲取數據庫後寫入緩存");
                    // 從數據庫查詢數據
                    result = getDataFromDB();
                    // 將查詢到的數據寫入緩存
                    setDataToCache(result);
                } finally {
                    reenLock.unlock();// 釋放鎖
                }
 
            } else {
                result = getDataFromCache();// 先查一下緩存
                if (result.isEmpty()) {
                    System.out.println("我沒拿到鎖,緩存也沒數據,先小憩一下");
                    Thread.sleep(100);// 小憩一下子
                    return getData04();// 重試
                }
            }
        }
        return result;
    }

 

注：最後使用互斥鎖的方式來實現，能夠有效避免前面幾種問題.

固然,在實際分佈式場景中，咱們還可使用 redis、tair、zookeeper 等提供的分佈式鎖來實現.可是,若是咱們的併發量若是隻有幾千的話,何須殺雞焉用牛刀呢?

轉自：https://www.cnblogs.com/dream-to-pku/p/9153999.html