緩存穿透bitmap

❶ SpringBoot+Redis BitMap 實現簽到與統計功能

歡迎來到技術小站，我們將持續分享優質的技術文章。

在項目開發中，簽到與統計功能的應用非常普遍。通過簽到，我們能夠為用戶提供獎勵，以此激勵用戶保持活躍，深入使用平台。

實現簽到功能，Redis中的BitMap是一個高效的選擇。相比於MySQL，BitMap能夠以極小的空間實現相同的功能。

BitMap的基本語法與指令如下：

Redis提供了一系列的BitMap操作指令，用於實現簽到功能。

為了實現簽到功能，我們將年和月作為BitMap的鍵，並將其存儲在BitMap中。每次簽到，就將對應的位從0變為1，表示用戶已簽到。

接下來，我們將通過SpringBoot整合Redis，實現簽到功能。

為了實現簽到功能，我們需要構建一個介面，用於將當前用戶當天的簽到信息保存至Redis中。

此外，我們還需要實現簽到統計功能，包括連續簽到天數、本月內的所有簽到數據以及連續簽到的次數等。

我們利用BitMap的特點，通過查詢BitMap的值，實現對連續簽到天數的統計。

同時，我們還需要關注緩存穿透問題的解決方案，通過將資料庫的數據對應的id寫入到一個list集合中，可以有效防止緩存穿透。

最後，通過整合SpringBoot與Redis，我們能夠實現高效的簽到與統計功能，提高用戶體驗，促進用戶活躍。

技術改變世界，希望這些知識能為你的項目帶來價值。

❷ 詳解布隆過濾器的原理和實現

為什麼需要布隆過濾器

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

手機號是否重復注冊

用戶是否參與過某秒殺活動

偽造請求大量 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 採用的就是此演算法：

//計算哈希次數@(longn,longm){//(m/n)*log(2),butavoidtruncationetodivision!returnMath.max(1,(int)Math.round((double)m/n*Math.log(2)));}//計算位數組長度@(longn,doublep){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=14type(//定義布隆過濾器結構體Filterstruct{bitsuintbitSetbitSetProvider}//位數組操作介面定義bitSetProviderinterface{check([]uint)(bool,error)set([]uint)error})

位數組操作介面實現

首先需要理解兩段 lua 腳本：

//ARGV:偏移量offset數組//KYES[1]:setbit操作的key//全部設置為1setScript=`for_,offsetinipairs(ARGV)doredis.call("setbit",KEYS[1],offset,1)end`//ARGV:偏移量offset數組//KYES[1]:setbit操作的key//檢查是否全部為1testScript=`for_,offsetinipairs(ARGV)doiftonumber(redis.call("getbit",KEYS[1],offset))==`

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

//redis位數組typeredisBitSetstruct{store*redis.Clientkeystringbitsuint}//檢查偏移量offset數組是否全部為1//是:元素可能存在//否:元素一定不存在func(r*redisBitSet)check(offsets[]uint)(bool,error){args,err:=r.buildOffsetArgs(offsets)iferr!=nil{returnfalse,err}//執行腳本resp,err:=r.store.Eval(testScript,[]string{r.key},args)//這里需要注意一下,底層使用的go-redis//redis.Nil表示key不存在的情況需特殊判斷iferr==redis.Nil{returnfalse,nil}elseiferr!=nil{returnfalse,err}exists,ok:=resp.(int64)if!ok{returnfalse,nil}returnexists==1,nil}//將k位點全部設置為1func(r*redisBitSet)set(offsets[]uint)error{args,err:=r.buildOffsetArgs(offsets)iferr!=nil{returnerr}_,err=r.store.Eval(setScript,[]string{r.key},args)//底層使用的是go-redis,redis.Nil表示操作的key不存在//需要針對key不存在的情況特殊判斷iferr==redis.Nil{returnnil}elseiferr!=nil{returnerr}returnnil}//構建偏移量offset字元串數組,因為go-redis執行lua腳本時參數定義為[]stringy//因此需要轉換一下func(r*redisBitSet)buildOffsetArgs(offsets[]uint)([]string,error){varargs[]stringfor_,offset:=rangeoffsets{ifoffset>=r.bits{returnnil,ErrTooLargeOffset}args=append(args,strconv.FormatUint(uint64(offset),10))}returnargs,nil}//刪除func(r*redisBitSet)del()error{_,err:=r.store.Del(r.key)returnerr}//自動過期func(r*redisBitSet)expire(secondsint)error{returnr.store.Expire(r.key,seconds)}funcnewRedisBitSet(store*redis.Client,keystring,bitsuint)*redisBitSet{return&redisBitSet{store:store,key:key,bits:bits,}}

到這里位數組操作就全部實現了，接下來看下如何通過 k 個散列函數計算出 k 個位點

k 次散列計算出 k 個位點

//k次散列計算出k個offsetfunc(f*Filter)getLocations(data[]byte)[]uint{//創建指定容量的切片locations:=make([]uint,maps)//maps表示k值,作者定義為了常量:14fori:=uint(0);i<maps;i++{//哈希計算,使用的是"MurmurHash3"演算法,並每次追加一個固定的i位元組進行計算hashValue:=hash.Hash(append(data,byte(i)))//取下標offsetlocations[i]=uint(hashValue%uint64(f.bits))}returnlocations}

插入與查詢

添加與查詢實現就非常簡單了，組合一下上面的函數就行。

//添加元素func(f*Filter)Add(data[]byte)error{locations:=f.getLocations(data)returnf.bitSet.set(locations)}//檢查是否存在func(f*Filter)Exists(data[]byte)(bool,error){locations:=f.getLocations(data)isSet,err:=f.bitSet.check(locations)iferr!=nil{returnfalse,err}if!isSet{returnfalse,nil}returntrue,nil}改進建議

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

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

預期總數量expectedInsertions

期望誤差falseProbability

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

//NewcreateaFilter,storeisthebackedredis,keyisthekeyforthebloomfilter,//bitsishowmanybitswillbeused,.//bestpractices://elements-meanshowmanyactualelements//whenmaps=14,formula:0.7*(bits/maps),bits=20*elements,theerrorrateis0.000067<1e-4//fordetailederrorratetable,seehttp://pages.cs.wisc.e/~cao/papers/summary-cache/node8.htmlfuncNew(store*redis.Redis,keystring,bitsuint)*Filter{return&Filter{bits:bits,bitSet:newRedisBitSet(store,key,bits),}}//expectedInsertions-預期總數量//falseProbability-預期誤差//這里也可以改為option模式不會破壞原有的兼容性funcNewFilter(store*redis.Redis,keystring,expectedInsertionsuint,falseProbabilityfloat64)*Filter{bits:=optimalNumOfBits(expectedInsertions,falseProbability)k:=optimalNumOfHashFunctions(bits,expectedInsertions)return&Filter{bits:bits,bitSet:newRedisBitSet(store,key,bits),k:k,}}//計算最優哈希次數funcoptimalNumOfHashFunctions(m,nuint)uint{returnuint(math.Round(float64(m)/float64(n)*math.Log(2)))}//計算最優數組長度funcoptimalNumOfBits(nuint,pfloat64)uint{returnuint(float64(-n)*math.Log(p)/(math.Log(2)*math.Log(2)))}回到問題

如何預防非法 id 導致緩存穿透?

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

採用布隆過濾器

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

緩存與資料庫同時無法命中時向緩存寫入一個過期時間較短的空值。

資料

布隆過濾器（Bloom Filter）原理及 Guava 中的具體實現

布隆過濾器-維基網路

Redis.setbit

項目地址

https://github.com/zeromicro/go-zero

歡迎使用 go-zero 並 star 支持我們！

微信交流群

關注『微服務實踐』公眾號並點擊 交流群 獲取社區群二維碼。

❸ 緩存擊穿互斥鎖 設置鎖的失效時間

• 設置鎖的失效時間是自己設置的，它的過期時間會很短，最長不超過五分鍾

• 緩存穿透是指查詢一個一定不存在的數據

• 由於緩存是不命中時被動寫的，

• 並且出於容錯考慮，如果從存儲層查不到數據則不寫入緩存，

• 這將導致這個不存在的數據每次請求都要到存儲層去查詢，失去了緩存的意義。

• 在流量大時，可能DB就掛掉了，

• 要是有人利用不存在的key頻繁攻擊我們的應用，這就是漏洞。

• 最常見的則是採用布隆過濾器

• 將所有可能存在的數據哈希到一個足夠大的bitmap中，

• 一個一定不存在的數據會被 這個bitmap攔截掉，

• 從而避免了對底層存儲系統的查詢壓力。

• 另外也有一個更為簡單粗暴的方法

• 如果一個查詢返回的數據為空（不管是數 據不存在，還是系統故障），

• 我們仍然把這個空結果進行緩存，但它的過期時間會很短，最長不超過五分鍾