Redis緩存穿透解決方法--布隆過濾器

　　

Redis的基於緩存，極大地提升了應用程序的性能和效率，特別是數據查詢方面，但是也帶來了一些問題，比如典型的

緩存穿透、緩存雪崩、緩存擊穿。

      緩存擊穿是指緩存中沒有但數據庫中有的數據（一般是緩存時間到期），這時由於並發用戶特別多，同時讀緩存沒讀到數據，又同時去數據庫去取數據，引起數據庫壓力瞬間增大，造成過大壓力

      解決方案：

1. 設置熱點數據永遠不過期。
2. 加互斥鎖

      緩存雪崩是指緩存中數據大批量到過期時間，而查詢數據量巨大，引起數據庫壓力過大甚至down機。和緩存擊穿不同的是，        緩存擊穿指並發查同一條數據，緩存雪崩是不同數據都過期了，很多數據都查不到從而查數據庫。

     解決方案：

1. 緩存數據的過期時間設置隨機，防止同一時間大量數據過期現象發生。
2. 如果緩存數據庫是分布式部署，將熱點數據均勻分布在不同搞得緩存數據庫中。
3. 設置熱點數據永遠不過期

　　緩存穿透（大量查詢一個不存在的key）定義：緩存穿透，是指查詢一個數據庫中不一定存在的數據；

　　正常使用緩存查詢數據的流程是，依據key去查詢value，數據查詢先進行緩存查詢，如果key不存在或者key已經過期，再對數據庫進行查詢，並把查詢到的對象，放進緩存。如果數據庫查詢對象為空，則不放進緩存。

如果每次都查詢一個不存在value的key，由於緩存中沒有數據，所以每次都會去查詢數據庫；當對key查詢的並發請求量很大時，每次都訪問DB，很可能對DB造成影響；並且由於緩存不命中，每次都查詢持久層，那么也失去了緩存的意義。

 　　解決方法

　　第一種是緩存層緩存空值

　　　　將數據庫中的空值也緩存到緩存層中，這樣查詢該空值就不會再訪問DB，而是直接在緩存層訪問就行。

　　　　但是這樣有個弊端就是緩存太多空值占用了更多的空間，可以通過給緩存層空值設立一個較短的過期時間來解決，例如60s。

　　第二種是布隆過濾器

　　　　將數據庫中所有的查詢條件，放入布隆過濾器中，

　　　　當一個查詢請求過來時，先經過布隆過濾器進行查，如果判斷請求查詢值存在，則繼續查；如果判斷請求查詢不存在，直接丟棄。

下面主要介紹布隆過濾器的使用。布隆過濾器介紹

布隆過濾器（Bloom Filter，下文簡稱BF）由Burton Howard Bloom在1970年提出，是一種空間效率高的概率型數據結構。它專門用來檢測集合中是否存在特定的元素。聽起來是很稀松平常的需求，為什么要使用BF這種數據結構呢？

設計思想

　　BF是由一個長度為m比特的位數組（bit array）與k個哈希函數（hash function）組成的數據結構。位數組均初始化為0，所有哈希函數都可以分別把輸入數據盡量均勻地散列。

當要插入一個元素時，將其數據分別輸入k個哈希函數，產生k個哈希值。以哈希值作為位數組中的下標，將所有k個對應的比特置為1。

　　當要查詢（即判斷是否存在）一個元素時，同樣將其數據輸入哈希函數，然后檢查對應的k個比特。如果有任意一個比特為0，表明該元素一定不在集合中。如果所有比特均為1，表明該集合有（較大的）可能性在集合中。為什么不是一定在集合中呢？因為一個比特被置為1有可能會受到其他元素的影響，這就是所謂“假陽性”（false positive）。相對地，“假陰性”（false negative）在BF中是絕不會出現的。

下圖示出一個m=18, k=3的BF示例。集合中的x、y、z三個元素通過3個不同的哈希函數散列到位數組中。當查詢元素w時，因為有一個比特為0，因此w不在該集合中。

 

 

優缺點與用途

BF的優點是顯而易見的：

• 不需要存儲數據本身，只用比特表示，因此空間占用相對於傳統方式有巨大的優勢，並且能夠保密數據；
• 時間效率也較高，插入和查詢的時間復雜度均為O(k)；
• 哈希函數之間相互獨立，可以在硬件指令層面並行計算。

但是，它的缺點也同樣明顯：

• 存在假陽性的概率，不適用於任何要求100%准確率的情境；
• 只能插入和查詢元素，不能刪除元素，這與產生假陽性的原因是相同的。我們可以簡單地想到通過計數（即將一個比特擴展為計數值）來記錄元素數，但仍然無法保證刪除的元素一定在集合中。

所以，BF在對查准度要求沒有那么苛刻，而對時間、空間效率要求較高的場合非常合適，本文第一句話提到的用途即屬於此類。另外，由於它不存在假陰性問題，所以用作“不存在”邏輯的處理時有奇效，比如可以用來作為緩存系統（如Redis）的緩沖，防止緩存穿透。

假陽性率的計算

假陽性是BF最大的痛點，因此有必要權衡，比如計算一下假陽性的概率。為了簡單一點，就假設我們的哈希函數選擇位數組中的比特時，都是等概率的。當然在設計哈希函數時，也應該盡量滿足均勻分布。

在位數組長度m的BF中插入一個元素，它的其中一個哈希函數會將某個特定的比特置為1。因此，在插入元素后，該比特仍然為0的概率是：

 

現有k個哈希函數，並插入n個元素，自然就可以得到該比特仍然為0的概率是：

 

 

 

 

反過來講，它已經被置為1的概率就是：

 

 

 

 

也就是說，如果在插入n個元素后，我們用一個不在集合中的元素來檢測，那么被誤報為存在於集合中的概率（也就是所有哈希函數對應的比特都為1的概率）為：

 

 

 

 

當n比較大時，根據重要極限公式，可以近似得出假陽性率：

 

 

所以，在哈希函數的個數k一定的情況下：

• 位數組長度m越大，假陽性率越低；
• 已插入元素的個數n越大，假陽性率越高。

事實上，即使哈希函數不是等概率選擇比特的，最終也會得出相同的結果，可以借助吾妻-霍夫丁不等式（Azuma-Hoeffding inequality）證明。我數學比較垃圾，就不班門弄斧了。

有一些框架內已經內建了BF的實現，免去了自己實現的煩惱。下面以Guava為例，看看Google是怎么做的。

Guava中的布隆過濾器

采用Guava 27.0.1版本的源碼，BF的具體邏輯位於com.google.common.hash.BloomFilter類中。開始讀代碼吧。

BloomFilter類的成員屬性

不多，只有4個。

  /** The bit set of the BloomFilter (not necessarily power of 2!) */
  private final LockFreeBitArray bits;

  /** Number of hashes per element */
  private final int numHashFunctions;

  /** The funnel to translate Ts to bytes */
  private final Funnel<? super T> funnel;

  /** The strategy we employ to map an element T to {@code numHashFunctions} bit indexes. */
  private final Strategy strategy;

 

• bits即上文講到的長度為m的位數組，采用LockFreeBitArray類型做了封裝。
• numHashFunctions即哈希函數的個數k。
• funnel是Funnel接口實現類的實例，它用於將任意類型T的輸入數據轉化為Java基本類型的數據（byte、int、char等等）。這里是會轉化為byte。
• strategy是布隆過濾器的哈希策略，即數據如何映射到位數組，其具體方法在BloomFilterStrategies枚舉中。

BloomFilter的構造

這個類的構造方法是私有的。要創建它的實例，應該通過公有的create()方法。它一共有5種重載方法，但最終都是調用了如下的邏輯

 @VisibleForTesting
  static <T> BloomFilter<T> create(
      Funnel<? super T> funnel, long expectedInsertions, double fpp, Strategy strategy) {
    checkNotNull(funnel);
    checkArgument(
        expectedInsertions >= 0, "Expected insertions (%s) must be >= 0", expectedInsertions);
    checkArgument(fpp > 0.0, "False positive probability (%s) must be > 0.0", fpp);
    checkArgument(fpp < 1.0, "False positive probability (%s) must be < 1.0", fpp);
    checkNotNull(strategy);

    if (expectedInsertions == 0) {
      expectedInsertions = 1;
    }
    /*
     * TODO(user): Put a warning in the javadoc about tiny fpp values, since the resulting size
     * is proportional to -log(p), but there is not much of a point after all, e.g.
     * optimalM(1000, 0.0000000000000001) = 76680 which is less than 10kb. Who cares!
     */
    long numBits = optimalNumOfBits(expectedInsertions, fpp);
    int numHashFunctions = optimalNumOfHashFunctions(expectedInsertions, numBits);
    try {
      return new BloomFilter<T>(new LockFreeBitArray(numBits), numHashFunctions, funnel, strategy);
    } catch (IllegalArgumentException e) {
      throw new IllegalArgumentException("Could not create BloomFilter of " + numBits + " bits", e);
    }
  }

 

該方法接受4個參數：funnel是插入數據的Funnel，expectedInsertions是期望插入的元素總個數n，fpp即期望假陽性率p，strategy即哈希策略。

由上可知，位數組的長度m和哈希函數的個數k分別通過optimalNumOfBits()方法和optimalNumOfHashFunctions()方法來估計。

估計最優m值和k值

  @VisibleForTesting
  static long optimalNumOfBits(long n, double p) {
    if (p == 0) {
      p = Double.MIN_VALUE;
    }
    return (long) (-n * Math.log(p) / (Math.log(2) * Math.log(2)));
  }

  @VisibleForTesting
  static int optimalNumOfHashFunctions(long n, long m) {
    // (m / n) * log(2), but avoid truncation due to division!
    return Math.max(1, (int) Math.round((double) m / n * Math.log(2)));
  }

 

由假陽性率的近似計算方法可知，如果要使假陽性率盡量小，在m和n給定的情況下，k值應為：

 

 

這就是optimalNumOfHashFunctions()方法的邏輯。那么m該如何估計呢？

將k代入上一節的式子並化簡，我們可以整理出期望假陽性率p與m、n的關系：

 

 

 

 

亦即：

 

 

這就是optimalNumOfBits()方法的邏輯。

從上也可以得出：

• 如果指定期望假陽性率p，那么最優的m值與期望元素數n呈線性關系。

•  最優的k值實際上只與p有關，與m和n都無關，即：

   

  

   

所以，在創建BloomFilter時，確定合適的p和n值很重要。

哈希策略

在BloomFilterStrategies枚舉中定義了兩種哈希策略，都基於著名的MurmurHash算法，分別是MURMUR128_MITZ_32和MURMUR128_MITZ_64。前者是一個簡化版，所以我們來看看后者的實現方法。

 

 MURMUR128_MITZ_64() {
    @Override
    public <T> boolean put(
        T object, Funnel<? super T> funnel, int numHashFunctions, LockFreeBitArray bits) {
      long bitSize = bits.bitSize();
      byte[] bytes = Hashing.murmur3_128().hashObject(object, funnel).getBytesInternal();
      long hash1 = lowerEight(bytes);
      long hash2 = upperEight(bytes);

      boolean bitsChanged = false;
      long combinedHash = hash1;
      for (int i = 0; i < numHashFunctions; i++) {
        // Make the combined hash positive and indexable
        bitsChanged |= bits.set((combinedHash & Long.MAX_VALUE) % bitSize);
        combinedHash += hash2;
      }
      return bitsChanged;
    }

    @Override
    public <T> boolean mightContain(
        T object, Funnel<? super T> funnel, int numHashFunctions, LockFreeBitArray bits) {
      long bitSize = bits.bitSize();
      byte[] bytes = Hashing.murmur3_128().hashObject(object, funnel).getBytesInternal();
      long hash1 = lowerEight(bytes);
      long hash2 = upperEight(bytes);

      long combinedHash = hash1;
      for (int i = 0; i < numHashFunctions; i++) {
        // Make the combined hash positive and indexable
        if (!bits.get((combinedHash & Long.MAX_VALUE) % bitSize)) {
          return false;
        }
        combinedHash += hash2;
      }
      return true;
    }

    private /* static */ long lowerEight(byte[] bytes) {
      return Longs.fromBytes(
          bytes[7], bytes[6], bytes[5], bytes[4], bytes[3], bytes[2], bytes[1], bytes[0]);
    }

    private /* static */ long upperEight(byte[] bytes) {
      return Longs.fromBytes(
          bytes[15], bytes[14], bytes[13], bytes[12], bytes[11], bytes[10], bytes[9], bytes[8]);
    }
  };

 

其中put()方法負責向布隆過濾器中插入元素，mightContain()方法負責判斷元素是否存在。以put()方法為例講解一下流程吧。

1. 使用MurmurHash算法對funnel的輸入數據進行散列，得到128bit（16B）的字節數組。
2. 取低8字節作為第一個哈希值hash1，取高8字節作為第二個哈希值hash2。
3. 進行k次循環，每次循環都用hash1與hash2的復合哈希做散列，然后對m取模，將位數組中的對應比特設為1。

這里需要注意兩點： 

 

在循環中實際上應用了雙重哈希（double hashing）的思想，即可以用兩個哈希函數來模擬k個，其中i為步長：

 

 

這種方法在開放定址的哈希表中，也經常用來減少沖突。

• 哈希值有可能為負數，而負數是不能在位數組中定位的。所以哈希值需要與Long.MAX_VALUE做bitwise AND，直接將其最高位（符號位）置為0，就變成正數了。

位數組具體實現

來看LockFreeBitArray類的部分代碼。

  static final class LockFreeBitArray {
    private static final int LONG_ADDRESSABLE_BITS = 6;
    final AtomicLongArray data;
    private final LongAddable bitCount;

    LockFreeBitArray(long bits) {
      this(new long[Ints.checkedCast(LongMath.divide(bits, 64, RoundingMode.CEILING))]);
    }

    // Used by serialization
    LockFreeBitArray(long[] data) {
      checkArgument(data.length > 0, "data length is zero!");
      this.data = new AtomicLongArray(data);
      this.bitCount = LongAddables.create();
      long bitCount = 0;
      for (long value : data) {
        bitCount += Long.bitCount(value);
      }
      this.bitCount.add(bitCount);
    }

    /** Returns true if the bit changed value. */
    boolean set(long bitIndex) {
      if (get(bitIndex)) {
        return false;
      }

      int longIndex = (int) (bitIndex >>> LONG_ADDRESSABLE_BITS);
      long mask = 1L << bitIndex; // only cares about low 6 bits of bitIndex

      long oldValue;
      long newValue;
      do {
        oldValue = data.get(longIndex);
        newValue = oldValue | mask;
        if (oldValue == newValue) {
          return false;
        }
      } while (!data.compareAndSet(longIndex, oldValue, newValue));

      // We turned the bit on, so increment bitCount.
      bitCount.increment();
      return true;
    }

    boolean get(long bitIndex) {
      return (data.get((int) (bitIndex >>> 6)) & (1L << bitIndex)) != 0;
    }
    // ....
}

 

看官應該能明白為什么它要叫做“LockFree”BitArray了，因為它是采用原子類型AtomicLongArray作為位數組的存儲的，確實不需要加鎖。另外還有一個Guava中特有的LongAddable類型的計數器，用來統計置為1的比特數。

采用AtomicLongArray除了有並發上的優勢之外，更主要的是它可以表示非常長的位數組。一個長整型數占用64bit，因此data[0]可以代表第0~63bit，data[1]代表64~127bit，data[2]代表128~191bit……依次類推。這樣設計的話，將下標i無符號右移6位就可以獲得data數組中對應的位置，再在其基礎上左移i位就可以取得對應的比特了。

最后多嘴一句，上面的代碼中用到了Long.bitCount()方法計算long型二進制表示中1的數量，堪稱Java語言中最強的騷操作之一：

 public static int bitCount(long i) {
    // HD, Figure 5-14
    i = i - ((i >>> 1) & 0x5555555555555555L);
    i = (i & 0x3333333333333333L) + ((i >>> 2) & 0x3333333333333333L);
    i = (i + (i >>> 4)) & 0x0f0f0f0f0f0f0f0fL;
    i = i + (i >>> 8);
    i = i + (i >>> 16);
    i = i + (i >>> 32);
    return (int)i & 0x7f;
 }