快取淘汰策略
LRU 與 LFU 快取策略及其實現。
應用層快取
鑑於磁碟和記憶體讀寫的差異性,DB 中低頻寫、高頻讀的資料適合放入記憶體中,直接供應用層讀寫。在專案中讀取使用者資料時就使用到了 LRU,而非放到 Redis 中。
快取的 2 個基本實現
Set(key string, value interface) // 寫資料 Get(key string) interface{}// 讀資料
快取的 2 個特徵
- 命中率:即命中數 / 請求數,比值越高即表明快取使用率越高,快取更有效。
- 淘汰策略:記憶體空間是有限的,當快取資料佔滿記憶體後,若要快取新資料,則必須淘汰一部分 舊 資料。對於 舊 的概念,不同淘汰策略有不同原則。
下邊介紹兩種常用的淘汰演算法:LRU 與 LFU
LRU
縮寫:Least Recently Used( 最近 最久 使用),時間維度
原則:若資料在最近一段時間內都未使用(讀取或更新),則以後使用機率也很低,應被淘汰。
資料結構
- 使用連結串列:由於快取讀寫刪都是高頻操作,考慮使用寫刪都為 O(1) 的 連結串列 ,而非寫刪都為 O(N) 的陣列。
- 使用雙鏈表:選用刪除操作為 O(1) 的 雙鏈表 而非刪除為 O(N) 的單鏈表。
-
維護額外雜湊表:連結串列查詢必須遍歷 O(N) 讀取,可在快取中維護
map[key]*Node
的 雜湊表 來實現O(1) 的連結串列查詢。
直接使用連結串列節點儲存快取的 K-V 資料,連結串列從 head 到 tail 使用頻率逐步降低。新訪問資料不斷追加到 head 前邊,舊資料不斷從 tail 剔除。LRU 使用連結串列順序性保證了熱資料在 head,冷資料在 tail。
雙鏈表節點儲存 K-V 資料:
type Node struct { keystring // 淘汰 tail 時需在維護的雜湊表中刪除,不是冗餘儲存 valinterface{} prev, next *Node // 雙向指標 } type List struct { head, tail *Node sizeint // 快取空間大小 }
從上圖可知,雙鏈表需實現快取節點新增 Prepend
,剔除 Remove
操作:
func (l *List) Prepend(node *Node) *Node { if l.head == nil { l.head = node l.tail = node } else { node.prev = nil node.next = l.head l.head.prev = node l.head = node } l.size++ return node } func (l *List) Remove(node *Node) *Node { if node == nil { return nil } prev, next := node.prev, node.next if prev == nil { l.head = next // 刪除頭結點 } else { prev.next = next } if next == nil { l.tail = prev // 刪除尾結點 } else { next.prev = prev } l.size-- node.prev, node.next = nil, nil return node } // 封裝資料已存在快取的後續操作 func (l *List) MoveToHead(node *Node) *Node { if node == nil { return nil } n := l.Remove(node) return l.Prepend(n) } func (l *List) Tail() *Node { return l.tail } func (l *List) Size() int { return l.size }
LRU 操作細節
Set(k, v)
- 資料已快取,則更新值,挪到 head 前
- 資料未快取
- 快取空間未滿:直接挪到 head 前
- 快取空間已滿:移除 tail 並將新資料挪到 head 前
Get(k)
- 命中:節點挪到 head 前,並返回 value
- 未命中:返回 -1
程式碼實現:
type LRUCache struct { capacity int // 快取空間大小 itemsmap[string]*Node list*List } func NewLRUCache(capacity int) *LRUCache { return &LRUCache{ capacity: capacity, items:make(map[string]*Node), list:new(List), } } func (c *LRUCache) Set(k string, v interface{}) { // 命中 if node, ok := c.items[k]; ok { node.val = v// 命中後更新值 c.items[k] = c.list.MoveToHead(node) // return } // 未命中 node := &Node{key: k, val: v} // 完整的 node if c.capacity == c.list.size { tail := c.list.Tail() delete(c.items, tail.key) // k-v 資料儲存與 node 中 c.list.Remove(tail) } c.items[k] = c.list.Prepend(node) // 更新地址 } func (c *LRUCache) Get(k string) interface{} { node, ok := c.items[k] if ok { c.items[k] = c.list.MoveToHead(node) return node.val } return -1 }
測試
func TestLRU(t *testing.T) { c := NewLRUCache(2) c.Set(K1, 1) c.Set(K2, 2) c.Set(K1, 100) fmt.Println(c.Get(K1)) // 100 c.Set(K3, 3) fmt.Println(c.Get(K2)) // -1 c.Set(K4, 4) fmt.Println(c.Get(K1)) // -1 fmt.Println(c.Get(K3)) // 3 fmt.Println(c.Get(K4)) // 4 }
LFU
縮寫:Least Frequently Used(最近 最少 使用),頻率維度。
原則:若資料在最近一段時間內使用次數少,則以後使用機率也很低,應被淘汰。
對比 LRU,若快取空間為 3 個數據量:
Set("2", 2) Set("1", 1) Get(1) Get(2) Set("3", 3) Set("4", 4) // LRU 將淘汰 1,快取連結串列為 4->3->2 // LFU 將淘汰 3,未超出容量的時段內 1 和 2 都被使用了兩次,3 僅使用一次
資料結構
依舊使用雙向連結串列實現高效寫刪操作,但 LFU 淘汰原則是 使用次數 ,資料節點在連結串列中的位置與之無關。可按使用次數劃分 頻率梯隊 ,資料節點使用一次就挪到高頻梯隊。此外維護 minFreq
表示最低梯隊,維護 2 個雜湊表:
map[freq]*List map[key]*Node
雙鏈表儲存快取資料:
type Node struct { keystring valinterface{} freqint // 將節點從舊梯隊移除時使用,非冗餘儲存 prev, next *Node } type List struct { head, tail *Node sizeint }
LFU 操作細節
Set(k, v)
- 資料已快取,則更新值,挪到下一梯隊
- 資料未快取
- 快取空間未滿:直接挪到第 1 梯隊
- 快取空間已滿:移除 minFreq 梯隊的 tail 節點 ,並將新資料挪到第 1 梯隊
Get(k)
- 命中:節點挪到下一梯隊,並返回 value
- 未命中:返回 -1
如上的 5 種 case,都要維護好對 minFreq
和 2 個雜湊表的讀寫。
程式碼實現:
type LFUCache struct { capacity int minFreqint // 最低頻率 items map[string]*Node freqs map[int]*List // 不同頻率梯隊 } func NewLFUCache(capacity int) *LFUCache { return &LFUCache{ capacity: capacity, minFreq:0, items:make(map[string]*Node), freqs:make(map[int]*List), } } func (c *LFUCache) Get(k string) interface{} { node, ok := c.items[k] if !ok { return -1 } // 移到 +1 梯隊中 c.freqs[node.freq].Remove(node) node.freq++ if _, ok := c.freqs[node.freq]; !ok { c.freqs[node.freq] = NewList() } newNode := c.freqs[node.freq].Prepend(node) c.items[k] = newNode // 新地址更新到 map if c.freqs[c.minFreq].Size() == 0 { c.minFreq++ // Get 的正好是當前值 } return newNode.val } func (c *LFUCache) Set(k string, v interface{}) { if c.capacity <= 0 { return } // 命中,需要更新頻率 if val := c.Get(k); val != -1 { c.items[k].val = v // 直接更新值即可 return } node := &Node{key: k, val: v, freq: 1} // 未命中 // 快取已滿 if c.capacity == len(c.items) { old := c.freqs[c.minFreq].Tail() // 最低最舊 c.freqs[c.minFreq].Remove(old) delete(c.items, old.key) } // 快取未滿,放入第 1 梯隊 c.items[k] = node if _, ok := c.freqs[1]; !ok { c.freqs[1] = NewList() } c.freqs[1].Prepend(node) c.minFreq = 1 }
minFreq
和 2 個雜湊表的維護使 LFU 比 LRU 更難實現。
測試
func TestLFU(t *testing.T) { c := NewLFUCache(2) c.Set(K1, 1)// 1:K1 c.Set(K2, 2)// 1:K2->K1 fmt.Println(c.Get(K1)) // 1:K2 2:K1 // 1 c.Set(K3, 3)// 1:K3 2:K1 fmt.Println(c.Get(K2)) // -1 fmt.Println(c.Get(K3)) // 2:k3->k1// 3 c.Set(K4, 4)// 1:K4 2:K3 fmt.Println(c.Get(K1)) // -1 fmt.Println(c.Get(K3)) // 1:K4 3:K3 // 3 }
總結
常見的快取淘汰策略有佇列直接實現的 FIFO,雙鏈表實現的 LFU 與 LRU,此外還有擴充套件的 2LRU 與 ARC 等演算法,它們的實現不依賴於任意一種資料結構,此外對於舊資料的衡量原則不同,淘汰策略也不一樣。
在演算法直接實現難度較大的情況下,不妨採用空間換時間,或時間換空間的策略來間接實現。要充分利用各種資料結構的優點並互補,比如連結串列加雜湊表就實現了任意操作 O(1) 複雜度的複合資料結構。