為什麼需要布隆過濾器

　　想象一下遇到下面的場景你會如何處理：

　　手機號是否重複註冊

　　使用者是否參與過某秒殺活動

　　偽造請求大量 id 查詢不存在的記錄，此時快取未命中,如何避免快取穿透

　　針對以上問題常規做法是：查詢資料庫，資料庫硬扛，如果壓力並不大可以使用此方法，保持簡單即可。

　　改進做法：用 list/set/tree 維護一個元素集合，判斷元素是否在集合內，時間複雜度或空間複雜度會比較高。如果是微服務的話可以用 redis 中的 list/set 資料結構, 資料規模非常大此方案的記憶體容量要求可能會非常高。

　　這些場景有個共同點，可以將問題抽象為：如何高效判斷一個元素不在集合中？ 那麼有沒有一種更好方案能達到時間複雜度和空間複雜雙優呢？

　　有！布隆過濾器。

　　什麼是布隆過濾器

　　布隆過濾器（英語：Bloom Filter）是 1970 年由布隆提出的。它實際上是一個很長的二進位制向量和一系列隨機對映函式。布隆過濾器可以用於檢索一個元素是否在一個集合中，它的優點是空間效率和查詢時間都遠遠超過一般的演算法。

　　> 工作原理

　　布隆過濾器的原理是，當一個元素被加入集合時，通過 K 個雜湊函式將這個元素對映成一個位數組中的 K 個點（offset），把它們置為 1。檢索時，我們只要看看這些點是不是都是 1 就（大約）知道集合中有沒有它了：如果這些點有任何一個 0，則被檢元素一定不在；如果都是 1，則被檢元素很可能在。這就是布隆過濾器的基本思想。

　　簡單來說就是準備一個長度為 m 的位陣列並初始化所有元素為 0，用 k 個雜湊函式對元素進行 k 次雜湊運算跟 len(m)取餘得到 k 個位置並將 m 中對應位置設定為 1。

　　布隆過濾器優缺點

　　優點：

　　空間佔用極小，因為本身不儲存資料而是用位元位表示資料是否存在，某種程度有保密的效果。

　　插入與查詢時間複雜度均為 O(k)，常數級別，k 表示雜湊函式執行次數。

　　雜湊函式之間可以相互獨立，可以在硬體指令層加速計算。

　　缺點：

　　誤差（假陽性率）。

　　無法刪除。

　　> 誤差（假陽性率）

　　布隆過濾器可以 100% 判斷元素不在集合中，但是當元素在集合中時可能存在誤判，因為當元素非常多時雜湊函式產生的 k 位點可能會重複。 維基百科有關於假陽性率的數學推導（見文末連結）這裡我們直接給結論(實際上是我沒看懂...)，假設：

　　位陣列長度 m

　　雜湊函式個數 k

　　預期元素數量 n

　　期望誤差_ε_

　　在建立布隆過濾器時我們為了找到合適的 m 和 k ，可以根據預期元素數量 n 與 ε 來推匯出最合適的 m 與 k 。

　　java 中 Guava, Redisson 實現布隆過濾器估算最優 m 和 k 採用的就是此演算法：

　　// 計算雜湊次數

　　@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)));

　　}

　　// 計算位陣列長度

　　@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)));

　　}

　　> 無法刪除

　　位陣列中的某些 k 點是多個元素重複使用的，假如我們將其中一個元素的 k 點全部置為 0 則直接就會影響其他元素。 這導致我們在使用布隆過濾器時無法處理元素被刪除的場景。 ​

　　可以通過定時重建的方式清除髒資料。假如是通過 redis 來實現的話重建時不要直接刪除原有的 key，而是先生成好新的再通過 rename 命令即可，再刪除舊資料即可。

　　go-zero 中的 bloom filter 原始碼分析

　　core/bloom/bloom.go ​ 一個布隆過濾器具備兩個核心屬性：

　　位陣列：

　　雜湊函式

　　go-zero實現的bloom filter中位陣列採用的是Redis.bitmap，既然採用的是 redis 自然就支援分散式場景，雜湊函式採用的是MurmurHash3

　　> Redis.bitmap 為什麼可以作為位陣列呢？

　　Redis 中的並沒有單獨的 bitmap 資料結構，底層使用的是動態字串（SDS）實現，而 Redis 中的字串實際都是以二進位制儲存的。 a 的ASCII碼是 97，轉換為二進位制是：01100001，如果我們要將其轉換為b只需要進一位即可：01100010。下面通過Redis.setbit實現這個操作：

　　> set foo a

　　> OK

　　> get foo

　　> "a"

　　> setbit foo 6 1

　　> 0

　　> setbit foo 7 0

　　> 1

　　> get foo

　　> "b"

　　bitmap 底層使用的動態字串可以實現動態擴容，當 offset 到高位時其他位置 bitmap 將會自動補 0，最大支援 2^32-1 長度的位陣列（佔用記憶體 512M），需要注意的是分配大記憶體會阻塞Redis程序。 根據上面的演算法原理可以知道實現布隆過濾器主要做三件事情：

　　k 次雜湊函式計算出 k 個位點。

　　插入時將位陣列中 k 個位點的值設定為 1。

　　查詢時根據 1 的計算結果判斷 k 位點是否全部為 1，否則表示該元素一定不存在。

　　下面來看看go-zero 是如何實現的：

　　> 物件定義

　　// 表示經過多少雜湊函式計算

　　// 固定14次

　　maps = 14

　　type (

　　// 定義布隆過濾器結構體

　　Filter struct {

　　bits uint

　　bitSet bitSetProvider

　　}

　　// 位陣列操作介面定義

　　bitSetProvider interface {

　　check([]uint) (bool, error)

　　set([]uint) error

　　}

　　)

　　> 位陣列操作介面實現

　　首先需要理解兩段 lua 指令碼：

　　// ARGV:偏移量offset陣列

　　// KYES[1]: setbit操作的key

　　// 全部設定為1

　　setScript = `

　　for _, offset in ipairs(ARGV) do

　　redis.call("setbit", KEYS[1], offset, 1)

　　end

　　`

　　// ARGV:偏移量offset陣列

　　// KYES[1]: setbit操作的key

　　// 檢查是否全部為1

　　testScript = `

　　for _, offset in ipairs(ARGV) do

　　if tonumber(redis.call("getbit", KEYS[1], offset)) == 0 then

　　return false

　　end

　　end

　　return true

　　`

　　為什麼一定要用 lua 指令碼呢? 因為需要保證整個操作是原子性執行的。

　　// redis位陣列

　　type redisBitSet struct {

　　store *redis.Client

　　key string

　　bits uint

　　}

　　// 檢查偏移量offset陣列是否全部為1

　　// 是:元素可能存在

　　// 否:元素一定不存在

　　func (r *redisBitSet) check(offsets []uint) (bool, error) {

　　args, err := r.buildOffsetArgs(offsets)

　　if err != nil {

　　return false, err

　　}

　　// 執行指令碼

　　resp, err := r.store.Eval(testScript, []string{r.key}, args)

　　// 這裡需要注意一下,底層使用的go-redis

　　// redis.Nil表示key不存在的情況需特殊判斷

　　if err == redis.Nil {

　　return false, nil

　　} else if err != nil {

　　return false, err

　　}

　　exists, ok := resp.(int64)

　　if !ok {

　　return false, nil

　　}

　　return exists == 1, nil

　　}

　　// 將k位點全部設定為1

　　func (r *redisBitSet) set(offsets []uint) error {

　　args, err := r.buildOffsetArgs(offsets)

　　if err != nil {

　　return err

　　}

　　_, err = r.store.Eval(setScript, []string{r.key}, args)

　　// 底層使用的是go-redis,redis.Nil表示操作的key不存在

　　// 需要針對key不存在的情況特殊判斷

　　if err == redis.Nil {

　　return nil

　　} Set) buildOffsetArgs(offsets []uint) ([]string, error) {

　　var args []string

　　for _, offset := range offsets {

　　if offset >= r.bits {

　　return nil, ErrTooLargeOffset

　　}

　　args = append(args, strconv.FormatUint(uint64(offset), 10))

　　}

　　return args, nil

　　}

　　// 刪除

　　func (r *redisBitSet) del() error {

　　_, err := r.store.Del(r.key)

　　return err

　　}

　　// 自動過期

　　func (r *redisBitSet) expire(seconds int) error {

　　return r.store.Expire(r.key, seconds)

　　}

　　func newRedisBitSet(store *redis.Client, key string, bits uint) *redisBitSet {

　　return &redisBitSet{

　　store: store,

　　key: key,

　　bits: bits,

　　}

　　}

　　到這裡位陣列操作就全部實現了，接下來看下如何通過 k 個雜湊函式計算出 k 個位點

　　> k 次雜湊計算出 k 個位點

　　// k次雜湊計算出k個offset

　　func (f *Filter) getLocations(data []byte) []uint {

　　// 建立指定容量的切片

　　locations := make([]uint, maps)

　　// maps表示k值,作者定義為了常量:14

　　for i := uint(0); i < maps; i++ {

　　// 雜湊計算,使用的是"MurmurHash3"演算法,並每次追加一個固定的i位元組進行計算

　　hashValue := hash.Hash(append(data, byte(i)))

　　// 取下標offset

　　locations[i] = uint(hashValue % uint64(f.bits))

　　}

　　return locations

　　}

　　> 插入與查詢

　　新增與查詢實現就非常簡單了，組合一下上面的函式就行。

　　// 新增元素

　　func (f *Filter) Add(data []byte) error {

　　locations := f.getLocations(data)

　　return f.bitSet.set(locations)

　　}

　　// 檢查是否存在

　　func (f *Filter) Exists(data []byte) (bool, error) {

　　locations := f.getLocations(data)

　　isSet, err := f.bitSet.check(locations)

　　if err != nil {

　　return false, err

　　}

　　if !isSet {

　　return false, nil

　　}

　　return true, nil

　　}

　　改進建議

　　整體實現非常簡潔高效，那麼有沒有改進的空間呢？

　　個人認為還是有的，上面提到過自動計算最優 m 與 k 的數學公式，如果建立引數改為：

　　預期總數量expectedInsertions

　　期望誤差falseProbability

　　就更好了，雖然作者註釋裡特別提到了誤差說明，但是實際上作為很多開發者對位陣列長度並不敏感，無法直觀知道 bits 傳多少預期誤差會是多少。

　　// New create a Filter, store is the backed redis, key is the key for the bloom filter,

　　// bits is how many bits will be used, maps is how many hashes for each addition.

　　// best practices:

　　// elements - means how many actual elements

　　// when maps = 14, formula: 0.7*(bits/maps), bits = 20*elements, the error rate is 0.000067 < 1e-4

　　// for detailed error rate table, see http://pages.cs.wisc.edu/~cao/papers/summary-cache/node8.html

　　func New(store *redis.Redis, key string, bits uint) *Filter {

　　return &Filter{

　　bits: bits,

　　bitSet: newRedisBitSet(store, key, bits),

　　}

　　}

　　// expectedInsertions - 預期總數量

　　// falseProbability - 預期誤差

　　// 這裡也可以改為option模式不會破壞原有的相容性

　　func NewFilter(store *redis.Redis, key string, expectedInsertions uint, falseProbability float64) *Filter {

　　bits := optimalNumOfBits(expectedInsertions, falseProbability)

　　k := optimalNumOfHashFunctions(bits, expectedInsertions)

　　return &Filter{

　　bits: bits,

　　bitSet: newRedisBitSet(store, key, bits),

　　k: k,

　　}

　　}

　　// 計算最優雜湊次數

　　func optimalNumOfHashFunctions(m, n uint) uint {

　　return uint(math.Round(float64(m) / float64(n) * math.Log(2)))

　　}

　　// 計算最優陣列長度

　　func optimalNumOfBits(n uint, p float64) uint {

　　return uint(float64(-n) * math.Log(p) / (math.Log(2) * math.Log(2)))

　　}

　　回到問題

　　> 如何預防非法 id 導致快取穿透?

　　由於 id 不存在導致請求無法命中快取流量直接打到資料庫，同時資料庫也不存在該記錄導致無法寫入快取，高併發場景這無疑會極大增加資料庫壓力。 解決方案有兩種：

　　採用布隆過濾器

　　資料寫入資料庫時需同步寫入布隆過濾器，同時如果存在髒資料場景（比如：刪除）則需要定時重建布隆過濾器，使用 redis 作為儲存時不可以直接刪除 bloom.key，可以採用 rename key 的方式更新 bloom

　　快取與資料庫同時無法命中時向快取寫入一個過期時間較短的空值。else if err != nil {

　　return err

　　}

　　return nil

　　}

　　// 構建偏移量offset字串陣列,因為go-redis執行lua指令碼時引數定義為[]stringy

　　// 因此需要轉換一下

　　func (r *redisBi