看完這篇Redis緩存三大問題，保你面試能造火箭，工做能擰螺絲。

時間 2020-06-09

標籤看完 redis 緩存三大問題面試火箭螺絲欄目 Redis 简体版

原文原文鏈接

前言

平常的開發中，無不都是使用數據庫來進行數據的存儲，因爲通常的系統任務中一般不會存在高併發的狀況，因此這樣看起來並無什麼問題。web

一旦涉及大數據量的需求，如一些商品搶購的情景，或者主頁訪問量瞬間較大的時候，單一使用數據庫來保存數據的系統會由於面向磁盤，磁盤讀/寫速度問題有嚴重的性能弊端，詳細的磁盤讀寫原理請參考這一片[]。面試

在這一瞬間成千上萬的請求到來，須要系統在極短的時間內完成成千上萬次的讀/寫操做，這個時候每每不是數據庫可以承受的，極其容易形成數據庫系統癱瘓，最終致使服務宕機的嚴重生產問題。redis

爲了克服上述的問題，項目一般會引入NoSQL技術，這是一種基於內存的數據庫，而且提供必定的持久化功能。數據庫

Redis技術就是NoSQL技術中的一種。Redis緩存的使用，極大的提高了應用程序的性能和效率，特別是數據查詢方面。api

但同時，它也帶來了一些問題。其中，最要害的問題，就是數據的一致性問題，從嚴格意義上講，這個問題無解。若是對數據的一致性要求很高，那麼就不能使用緩存。數組

另外的一些典型問題就是，緩存穿透、緩存擊穿和緩存雪崩。本篇文章從實際代碼操做，來提出解決這三個緩存問題的方案，畢竟Redis的緩存問題是實際面試中高頻問點，理論和實操要兼得。緩存

緩存穿透

緩存穿透是指查詢一條數據庫和緩存都沒有的一條數據，就會一直查詢數據庫，對數據庫的訪問壓力就會增大，緩存穿透的解決方案，有如下兩種：安全

緩存空對象：代碼維護較簡單，可是效果很差。
布隆過濾器：代碼維護複雜，效果很好。

緩存空對象

緩存空對象是指當一個請求過來緩存中和數據庫中都不存在該請求的數據，第一次請求就會跳過緩存進行數據庫的訪問，而且訪問數據庫後返回爲空，此時也將該空對象進行緩存。數據結構

如果再次進行訪問該空對象的時候，就會直接擊中緩存，而不是再次數據庫，緩存空對象實現的原理圖以下：緩存空對象的實現代碼以下：併發

public class UserServiceImpl {
     @Autowired
     UserDAO userDAO;
     @Autowired
     RedisCache redisCache;
 public User findUser(Integer id) {
      Object object = redisCache.get(Integer.toString(id));
      // 緩存中存在，直接返回
      if(object != null) {
           // 檢驗該對象是否爲緩存空對象，是則直接返回null
           if(object instanceof NullValueResultDO) {
                return null;
           }
           return (User)object;
      } else {  
           // 緩存中不存在，查詢數據庫
           User user = userDAO.getUser(id);
           // 存入緩存
           if(user != null) {
                redisCache.put(Integer.toString(id),user);
           } else {
                // 將空對象存進緩存
                redisCache.put(Integer.toString(id), new NullValueResultDO());
           }
           return user;
      }
 }          
複製代碼
複製代碼public User findUser(Integer id) {
      Object object = redisCache.get(Integer.toString(id));
      // 緩存中存在，直接返回
      if(object != null) {
           // 檢驗該對象是否爲緩存空對象，是則直接返回null
           if(object instanceof NullValueResultDO) {
                return null;
           }
           return (User)object;
      } else {  
           // 緩存中不存在，查詢數據庫
           User user = userDAO.getUser(id);
           // 存入緩存
           if(user != null) {
                redisCache.put(Integer.toString(id),user);
           } else {
                // 將空對象存進緩存
                redisCache.put(Integer.toString(id), new NullValueResultDO());
           }
           return user;
      }
 }          
複製代碼} 複製代碼

緩存空對象的實現代碼很簡單，可是緩存空對象會帶來比較大的問題，就是緩存中會存在不少空對象，佔用內存的空間，浪費資源，一個解決的辦法就是設置空對象的較短的過時時間，代碼以下：

// 再緩存的時候，添加多一個該空對象的過時時間60秒
redisCache.put(Integer.toString(id), new NullValueResultDO(),60);
複製代碼

布隆過濾器

布隆過濾器是一種基於機率的數據結構，主要用來判斷某個元素是否在集合內，它具備運行速度快（時間效率），佔用內存小的優勢（空間效率），可是有必定的誤識別率和刪除困難的問題。它只能告訴你某個元素必定不在集合內或可能在集合內。

在計算機科學中有一種思想：空間換時間，時間換空間。通常二者是不可兼得，而布隆過濾器運行效率和空間大小都兼得，它是怎麼作到的呢？

在布隆過濾器中引用了一個誤判率的概念，即它可能會把不屬於這個集合的元素認爲可能屬於這個集合，可是不會把屬於這個集合的認爲不屬於這個集合，布隆過濾器的特色以下：

一個很是大 的二進制位數組 （數組裏只有0和1）
若干個 哈希函數
空間效率和 查詢效率高
不存在漏報（False Negative）：某個元素在某個集合中，確定能報出來。
可能存在誤報（False Positive）：某個元素不在某個集合中，可能也被爆出來。
不提供刪除方法，代碼維護困難。
位數組初始化都爲0，它不存元素的具體值，當元素通過哈希函數哈希後的值（也就是數組下標）對應的數組位置值改成1。

實際布隆過濾器存儲數據和查詢數據的原理圖以下：可能不少讀者看完上面的特色和原理圖，仍是看不懂，別急下面經過圖解一步一步的講解布隆過濾器，總而言之一句簡單的話歸納就是布隆過濾器是一個很大二進制的位數組，數組裏面只存0和1。

初始化的布隆過濾器的結構圖以下：以上只是畫了布隆過濾器的很小很小的一部分，實際布隆過濾器是很是大的數組（這裏的大是指它的長度大，並非指它所佔的內存空間大）。

那麼一個數據是怎麼存進布隆過濾器的呢？

當一個數據進行存入布隆過濾器的時候，會通過如干個哈希函數進行哈希（如果對哈希函數還不懂的請參考這一片[]），獲得對應的哈希值做爲數組的下標，而後將初始化的位數組對應的下標的值修改成1，結果圖以下：

當再次進行存入第二個值的時候，修改後的結果的原理圖以下：因此每次存入一個數據，就會哈希函數的計算，計算的結果就會做爲下標，在布隆過濾器中有多少個哈希函數就會計算出多少個下標，布隆過濾器插入的流程以下：

將要添加的元素給m個哈希函數
獲得對應於位數組上的m個位置
將這m個位置設爲1

那麼爲何會有誤判率呢？

假設在咱們屢次存入值後，在布隆過濾器中存在x、y、z這三個值，布隆過濾器的存儲結構圖以下所示：當咱們要查詢的時候，好比查詢a這個數，實際中a這個數是不存在布隆過濾器中的，通過2哥哈希函數計算後獲得a的哈希值分別爲2和13，結構原理圖以下：通過查詢後，發現2和13位置所存儲的值都爲1，可是2和13的下標分別是x和z通過計算後的下標位置的修改，該布隆過濾器中實際不存在a，那麼布隆過濾器就會誤判改值可能存在，由於布隆過濾器不存元素值，因此存在誤判率。

那麼具體布隆過布隆過濾的判斷的準確率和一下兩個因素有關：

布隆過濾器大小：越大，誤判率就越小，因此說布隆過濾器通常長度都是很是大的。
哈希函數的個數：哈希函數的個數越多，那麼誤判率就越小。

那麼爲何不能刪除元素呢？

緣由很簡單，由於刪除元素後，將對應元素的下標設置爲零，可能別的元素的下標也引用改下標，這樣別的元素的判斷就會收到影響，原理圖以下：當你刪除z元素以後，將對應的下標10和13設置爲0，這樣致使x和y元素的下標受到影響，致使數據的判斷不許確，因此直接不提供刪除元素的api。

以上說的都是布隆過濾器的原理，只有理解了原理，在實際的運用才能如魚得水，下面就來實操代碼，手寫一個簡單的布隆過濾器。

對於要手寫一個布隆過濾器，首先要明確布隆過濾器的核心：

若干哈希函數
存值得Api
判斷值得Api

實現得代碼以下：

public class MyBloomFilter {
    // 布隆過濾器長度
    private static final int SIZE = 2 << 10;
    // 模擬實現不一樣的哈希函數
    private static final int[] num= new int[] {5, 19, 23, 31,47, 71};   
    // 初始化位數組
    private BitSet bits = new BitSet(SIZE);
    // 用於存儲哈希函數
    private MyHash[] function = new MyHash[num.length];
// 初始化哈希函數
public MyBloomFilter() {
    for (int i = 0; i &lt; num.length; i++) {
        function [i] = new MyHash(SIZE, num[i]);
    }
}

// 存值Api 
public void add(String value) {
    // 對存入得值進行哈希計算
    for (MyHash f: function) {
        // 將爲數組對應的哈希下標得位置得值改成1
        bits.set(f.hash(value), true);
    }
}

// 判斷是否存在該值得Api 
public boolean contains(String value) {
    if (value == null) {
        return false;
    }
    boolean result= true;
    for (MyHash f : func) {
        result= result&amp;&amp; bits.get(f.hash(value));
    }
    return result;
}
複製代碼
複製代碼// 初始化哈希函數
public MyBloomFilter() {
    for (int i = 0; i &lt; num.length; i++) {
        function [i] = new MyHash(SIZE, num[i]);
    }
}

// 存值Api 
public void add(String value) {
    // 對存入得值進行哈希計算
    for (MyHash f: function) {
        // 將爲數組對應的哈希下標得位置得值改成1
        bits.set(f.hash(value), true);
    }
}

// 判斷是否存在該值得Api 
public boolean contains(String value) {
    if (value == null) {
        return false;
    }
    boolean result= true;
    for (MyHash f : func) {
        result= result&amp;&amp; bits.get(f.hash(value));
    }
    return result;
}
複製代碼} 複製代碼

哈希函數代碼以下：

public static class MyHash {
        private int cap;
        private int seed;
        // 初始化數據
        public MyHash(int cap, int seed) {
            this.cap = cap;
            this.seed = seed;
        }
        // 哈希函數
        public int hash(String value) {
            int result = 0;
            int len = value.length();
            for (int i = 0; i < len; i++) {
                result = seed * result + value.charAt(i);
            }
            return (cap - 1) & result;
        }
    }
複製代碼

布隆過濾器測試代碼以下：

public static void test {
        String value = "4243212355312";
        MyBloomFilter filter = new MyBloomFilter();
        System.out.println(filter.contains(value));
        filter.add(value);
        System.out.println(filter.contains(value));
    }
複製代碼

以上就是手寫了一個很是簡單得布隆過濾器，可是實際項目中可能事由牛人或者大公司已經幫你寫好的，如谷歌的Google Guava，只須要在項目中引入一下依賴：

<dependency>
    <groupId>com.google.guava</groupId>
    <artifactId>guava</artifactId>
    <version>27.0.1-jre</version>
</dependency>
複製代碼

實際項目中具體的操做代碼以下：

public static void MyBloomFilterSysConfig {
 @Autowired
 OrderMapper orderMapper
 
// 1.建立布隆過濾器  第二個參數爲預期數據量10000000，第三個參數爲錯誤率0.00001
BloomFilter&lt;CharSequence&gt; bloomFilter =  BloomFilter.create(Funnels.stringFunnel(Charset.forName("utf-8")),10000000, 0.00001);
// 2.獲取全部的訂單，並將訂單的id放進布隆過濾器裏面
List&lt;Order&gt; orderList = orderMapper.findAll()
for (Order order;orderList ) {
    Long id = order.getId();
    bloomFilter.put("" + id);
}
複製代碼
複製代碼@Autowired
 OrderMapper orderMapper
 
// 1.建立布隆過濾器  第二個參數爲預期數據量10000000，第三個參數爲錯誤率0.00001
BloomFilter&lt;CharSequence&gt; bloomFilter =  BloomFilter.create(Funnels.stringFunnel(Charset.forName("utf-8")),10000000, 0.00001);
// 2.獲取全部的訂單，並將訂單的id放進布隆過濾器裏面
List&lt;Order&gt; orderList = orderMapper.findAll()
for (Order order;orderList ) {
    Long id = order.getId();
    bloomFilter.put("" + id);
}
複製代碼} 複製代碼

在實際項目中會啓動一個系統任務或者定時任務，來初始化布隆過濾器，將熱點查詢數據的id放進布隆過濾器裏面，當用戶再次請求的時候，使用布隆過濾器進行判斷，改訂單的id是否在布隆過濾器中存在，不存在直接返回null，具體操做代碼：

// 判斷訂單id是否在布隆過濾器中存在
bloomFilter.mightContain("" + id)
複製代碼

布隆過濾器的缺點就是要維持容器中的數據，由於訂單數據確定是頻繁變化的，實時的要更新布隆過濾器中的數據爲最新。

緩存擊穿

緩存擊穿是指一個key很是熱點，在不停的扛着大併發，大併發集中對這一個點進行訪問，當這個key在失效的瞬間，持續的大併發就穿破緩存，直接請求數據庫，瞬間對數據庫的訪問壓力增大。

緩存擊穿這裏強調的是併發，形成緩存擊穿的緣由有如下兩個：

該數據沒有人查詢過，第一次就大併發的訪問。（冷門數據）
添加到了緩存，reids有設置數據失效的時間，這條數據恰好失效，大併發訪問（熱點數據）

對於緩存擊穿的解決方案就是加鎖，具體實現的原理圖以下：當用戶出現大併發訪問的時候，在查詢緩存的時候和查詢數據庫的過程加鎖，只能第一個進來的請求進行執行，當第一個請求把該數據放進緩存中，接下來的訪問就會直接集中緩存，防止了緩存擊穿。

業界比價廣泛的一種作法，即根據key獲取value值爲空時，鎖上，從數據庫中load數據後再釋放鎖。若其它線程獲取鎖失敗，則等待一段時間後重試。這裏要注意，分佈式環境中要使用分佈式鎖，單機的話用普通的鎖（synchronized、Lock）就夠了。

下面以一個獲取商品庫存的案例進行代碼的演示，單機版的鎖實現具體實現的代碼以下：

// 獲取庫存數量
public String getProduceNum(String key) {
    try {
        synchronized (this) {   //加鎖
            // 緩存中取數據，並存入緩存中
            int num= Integer.parseInt(redisTemplate.opsForValue().get(key));
        if (num&gt; 0) {
            //沒查一次庫存-1
            redisTemplate.opsForValue().set(key, (num- 1) + "");
            System.out.println("剩餘的庫存爲num：" + (num- 1));
        } else {
            System.out.println("庫存爲0");
        }
    }
} catch (NumberFormatException e) {
    e.printStackTrace();
} finally {
}
return "OK";
複製代碼
複製代碼if (num&gt; 0) {
            //沒查一次庫存-1
            redisTemplate.opsForValue().set(key, (num- 1) + "");
            System.out.println("剩餘的庫存爲num：" + (num- 1));
        } else {
            System.out.println("庫存爲0");
        }
    }
} catch (NumberFormatException e) {
    e.printStackTrace();
} finally {
}
return "OK";
複製代碼} 複製代碼

分佈式的鎖實現具體實現的代碼以下：

public String getProduceNum(String key) {
    // 獲取分佈式鎖
    RLock lock = redissonClient.getLock(key);
    try {
        // 獲取庫存數
        int num= Integer.parseInt(redisTemplate.opsForValue().get(key));  
        // 上鎖           
        lock.lock();
        if (num> 0) {
            //減小庫存，並存入緩存中
            redisTemplate.opsForValue().set(key, (num - 1) + "");
            System.out.println("剩餘庫存爲num：" + (num- 1));
        } else {
            System.out.println("庫存已經爲0");
        }
    } catch (NumberFormatException e) {
        e.printStackTrace();
    } finally {
        //解鎖
        lock.unlock();
    }
    return "OK";
}
複製代碼