文章目录
- LFU 算法
- 理论介绍
- 算法实现
- 数据结构
- 查询操作
- 插入/更新操作
- Redis 缓存淘汰算法
- 缓存污染难题
- Redis LFU缓存淘汰策略
本篇博客的主要内容:
- 以图解的方式,介绍 LFU 算法的一种实现;
- 介绍 LFU 算法在 Redis 中的应用。
LFU 算法
理论介绍
最不常使用 (LFU Least Frequently Used) 算法是一种常用的缓存置换算法,旨在移除 访问频次最低 的缓存项。与常用的 LRU 算法不同,LFU 算法更加重视元素的访问频率,而非最近一次访问时间。
实现 LFU 淘汰算法的缓存需要满足如下规则:
- 缓存中键值对的数目不能超过容量
capacity
。 - get(key) 操作:如果键 key 存在于缓存,则返回该键;否则空值。
- put(key, val) 操作:如果 key 已经存在,变更其值;如果 key 不存在,插入键值对。当缓存达到其容量 capacity 时,则应该在插入新项之前,移除 最不经常使用 的键值对。如果存在两个或更多键具有相同的最小频率,应该去除 最近最久未使用(LRU) 的键值对。
解释下高亮部分:
- 最不经常使用的键值对:使用频数最少的键,使用 get(key) 或 put(key, val) 访问键对都会增加键的频数。当一个键首次插入到缓存中时,它的使用计数器被设置为
1
。 - 最近最久未使用的键值对:缓存中键的最小频数为 count_min,当需要淘汰缓存时,可能存在多个键的频数等于 count_min。对于这种情况,针对频数等于 count_min 的键值对,采用 LRU(最近最久未使用)算法淘汰。
算法实现
本节介绍如何设计 get、put 函数以 O(1) 时间复杂度运行 的 LFU 缓存。
数据结构
缓存中的键值对通过一个全局双向循环链表存储,链表节点 Node 的定义如下:
struct Node{int key;int val;Node* prev;Node* next;int count;Node(int k, int v): key(k), val(v), prev(nullptr), next(nullptr), count(1) {}Node(int k, int v, Node *p, Node *n): key(k), val(v), prev(p), next(n), count(1) {}
};
- key、val:键值对的键与值
- prev:前驱节点;
- next:后继节点;
- count:节点的访问次数,新节点插入时设置初始值 1。
缓存中使用字段 head
存储双向循环链表的头节点,头节点是 链表中频数最高的节点:
Node *head = nullptr;
LFU 缓存是一个由多个 链表分段 拼装而成的双向循环链表,相同链表段中的节点频数相等。示意图如下:
当一个频数 count 的节点被访问,频数将递增为 count + 1,我们需要在 O(1) 的时间将该节点从频数 count 的链表分段移动到频数 count + 1 的链表分段头部。
因此,需要一个哈希表 count2SegHead
维护频数count 到链 表分段头节点 的映射:
unordered_map<int, Node*> count2SegHead;
示意图如下:
为了在 O(1) 时间复杂度通过 key 访问节点,需要使用哈希表 key2Node
维护 键key 到 节点 的映射:
unordered_map<int, Node*> key2Node;
示意图如下:
查询操作
查询操作步骤如下:
- 通过哈希表 key2node 查询到指定键的节点 node;
- 将 node 节点从双向循环链表中移除;
- 节点频数为 count,找到 count + 1 的链表分段头节点 seg_head;
- 更新
node.count = count + 1
,将 node 节点插入 seg_head 前,同时设置count2SegHead[count + 1] = node
,即设置 node 节点为频数 count + 1 的链表段头节点。
下面这副示意图展示了查询 key=8 节点的流程:
- 通过 key2Node 找到节点位置,key=8 节点的频数为 3。
- 从频数 3 的链表分段中移除 key=8 节点;
- 查询 count2SegHead 得到频数为 3+1=4 的链表分段头节点(key=3),将 key=8 节点的频数更新为 4,然后插入到 key=3 节点的前驱。
- 最后,更新 head 指针以及
count2SegHead[4] = node
;
查询函数 getNode
代码实现如下:
#include "bits/stdc++.h"using namespace std;class LFUCache {struct Node{int key;int val;Node* prev;Node* next;int count;Node(int k, int v): key(k), val(v), prev(nullptr), next(nullptr), count(1) {}Node(int k, int v, Node *p, Node *n): key(k), val(v), prev(p), next(n), count(1) {}};unordered_map<int, Node*> key2Node;unordered_map<int, Node*> count2SegHead; // 频数到链表分段的映射int capacity;int size = 0;Node *head = nullptr; // 双向循环链表头节点// 将链表中的 node 节点移动到 newCount 链表段void moveNode(Node* node, int newCount) { // ------(1)auto nextNode = remove(node);node->count = newCount;if(count2SegHead.find(newCount) == count2SegHead.end())insert(node, nextNode);elseinsert(node, count2SegHead[newCount]);}// 将节点插入到链表段头节点 segHead 前, 同时维护 head 指针和 count2LinkedListvoid insert(Node* node, Node* segHead) {count2SegHead[node->count] = node;if(segHead == nullptr) {node->next = node->prev = node;head = node;return;}node->next = segHead, node->prev = segHead->prev;node->prev->next = node, node->next->prev = node;if(head->count <= node->count) head = node;}// remove 函数Node* remove(Node *node){// 循环链表中仅有node节点if(node->next == node){count2SegHead.erase(node->count);head = nullptr;return nullptr;}Node* next;// node 节点为 node->count 链表段的头节点if(count2SegHead[node->count] == node) {if(node->next->count == node->count)count2SegHead[node->count] = node->next;elsecount2SegHead.erase(node->count);next = node->next;} // node 不是链表段头节点else next = count2SegHead[node->count];node->prev->next = node->next, node->next->prev = node->prev;node->prev = node->next = nullptr;if(head == node) head = next;return next;}public:LFUCache(int cap) {this->capacity = cap;}Node* getNode(int key){if(key2Node.find(key) == key2Node.end()) return nullptr;auto node = key2Node[key];moveNode(node, node->count + 1); // ------(1)return node;}
};
moveHead(Node* node, int newCount)
:将循环链表中的节点 node 移动到频数为 newCount 的链表段头部。
moveHead
执行 remove(Node *node)
函数从双向循环链表中移除 node 节点:
- 如果循环链表中只有 node 节点,将 head 设置为 nullptr,
count2SegHead
哈希表移除频数为 node->count 的键值对; - 如果 node 为
node->count
链表段的头节点,考察node->next
的频数:- 如果相等,将
node->next
设置为链表段新的头节点; - 如果不相等,说明 node 为该链表段的唯一节点,移除 node 后链表段也被移除,count2SegHead 移除
node->count
键值对。
- 如果相等,将
- 如果 node 不为链表段头节点,可以直接移除,无需维护 head 及 count2SegHead;
- 返回值:
node->count
链表段移除 node 后的头节点;如果链表段已移除,则返回该链表段的后继链表段头节点。
Node* remove(Node *node){// 循环链表中仅有node节点if(node->next == node){count2SegHead.erase(node->count);head = nullptr;return nullptr;}Node* next;// node 节点为 node->count 链表段的头节点if(count2SegHead[node->count] == node) {if(node->next->count == node->count)count2SegHead[node->count] = node->next;elsecount2SegHead.erase(node->count);next = node->next;} // node 不是链表段头节点else next = count2SegHead[node->count];node->prev->next = node->next, node->next->prev = node->prev;node->prev = node->next = nullptr;if(head == node) head = next;return next;
}
移除 node 节点后,使用 insert(Node* node, Node* segHead)
将 node 插入到链表段 segHead 的头部:
- 将 node 设置为
node->count
链表段新头节点:count2SegHead[node->count] = node
; - 如果 segHead 为 nullptr,说明全局链表为空,执行双向循环链表的初始化操作;
- 如果 segHead 非空,将 node 插入到 segHead 的前驱。
- 如果 node.count 操作计数大于等于当前头节点 head 的计数,将 node 设置为新的头节点。
// 将节点插入到链表段头节点 segHead 前, 同时维护 head 指针和 count2LinkedListvoid insert(Node* node, Node* segHead) {count2SegHead[node->count] = node; // 设置为新的链表段头节点if(segHead == nullptr) {// segHead 为 null, 说明全局链表为空, 执行初始化操作node->next = node->prev = node;head = node;return;}// 双向链表插入操作node->next = segHead, node->prev = segHead->prev;node->prev->next = node, node->next->prev = node;// 如果 node 操作计数大于等于当前头节点的计数, 更新 headif(head->count <= node->count) head = node;}
插入/更新操作
插入/更新操作的步骤如下:
- 使用
getNode
查询键为 key 的节点; - 如果节点存在,只需要更新 node.val 为 value 新值,频数更新、双向循环链表的维护由
getNode
函数完成。 - 如果节点不存在,则执行新节点插入操作,新节点的初始频数 count 等于 1;
- 缓存大小等于容量,则淘汰缓存中频数最少的节点,如果频数最少的节点非唯一,淘汰最近一次访问时间最早(LRU)的节点。
- 缓存大小小于容量,将节点插入到频数 1 的链表段头节点之前。
还需要考虑到,当前不存在频数为 1 的节点,也即不存在频数为 1 的链表段。这种情况下,将新节点插入到 head 节点之前即可。
void put(int key, int value) {if(capacity <= 0) return;Node *node = getNode(key);if(node == nullptr) {// 键等于 key 的节点不存在, 插入新的节点if(size == capacity){// 容量已经达到上限, 淘汰频数最低的节点, 即 head.prevauto evict = head->prev;remove(evict);key2Node.erase(evict->key);delete evict;} else size++;// 新的节点, node = new Node(key, value, nullptr, nullptr);key2Node[key] = node;if(count2SegHead.find(1) == count2SegHead.end())// 如果不存在频数为 1 的链表段, 将新节点插入到head的前驱;insert(node, head);else // 如果存在频数等于 1 的链表段, 新节点插入到该链表段的头节点前insert(node, count2SegHead[1]);} else node->val = value; // 更新键等于 key 的节点值
}
下面,我用两个场景,带大家理解 LFU 缓存的插入操作:
场景1:插入节点后缓存大小(size=7) 未超出 容量capacity(7),新节点(key=4) 插入到频数等于 1 的链表段前,成为该链表段的头节点。示意图如下:
场景2:插入节点后缓存大小(size=7) 超出 容量capacity(6),先 淘汰操作频数最小的节点 (key=7),此时频数为 1 的链表段将被移除,新节点(key=4) 插入到头节点(key=3) 之前。示意图如下:
Redis 缓存淘汰算法
Redis 中实现了基于 LFU 的缓存淘汰策略:volatile-lfu 和 allkeys-lfu。在介绍该策略之前,我们先了解下 缓存污染 的概念。
缓存污染难题
缓存污染 是指 访问很少的数据 在服务完访问请求后还继续 留存在缓存中,造成缓存空间的浪费。
缓存污染一旦变得严重后,有大量不再访问的数据滞留在缓存中,往缓存中写入新数据时需要先把数据逐步淘汰出缓存,引入额外的操作时间开销。
如何解决缓存污染问题?
- volatile-ttl 策略:如果业务应用在设置过期时间时,明确知道数据的访问情况(即数据的时效性),Redis 按照数据的剩余最短存活时间淘汰数据,可以避免缓存污染。
- volatile-lru、all-lru:淘汰候选数据集中,lru 字段最小(最近一次访问时间最久)的数据。
LRU 方案的缺陷:只考虑了数据的访问时间,没有考虑数据的访问频数,在处理 扫描式单次查询操作 时,无法解决缓存问题!
幸运的是,Redis 从 4.0 版本开始增加了 LFU 淘汰策略:从 数据访问的时效性 、数据访问次数 两个角度筛选出需要淘汰的数据。
Redis LFU缓存淘汰策略
在实现 LRU 算法时,Redis 使用 RedisObject
结构来保存数据的;该结构的 lru
字段记录数据最近一次访问的时间戳。实现 LFU 算法时,将原来 24bit 的 lru 字段,进一步拆分成两个部分:
- ldt 值:lru 字段的前 16bit,表示数据的访问时间戳;
- counter 值:lru 字段的后 8bit,表示数据的访问次数。
Redis使用LFU策略淘汰数据时,选择 lru 字段最小的数据进行淘汰。等价于 优先淘汰访问次数少的数据,访问次数相等则淘汰时间戳最小 的数据。
注意:count 为计数器的值,初始默认为 5 而不是 1。如果初始值为 1,刚被写入缓存的数据可能会因为使用次数太少而立即被淘汰。
counter 值 8bits,最大值 255,如果采用线性增加的方式 counter 很快就会达到上限,Redis 就不能很好筛选访问 1000次 和 10000 次的数据。因此,Redis 采用了非线性的频数更新策略:
p = 1 c o u n t × l f u _ l o g _ f a c t o r + 1 p=\frac{1}{count \times {lfu\_log\_factor} + 1} p=count×lfu_log_factor+11
每次缓存的数据被访问, counter 加 1 的概率等于 p p p,源码如下:
double r = (double)rand()/RAND_MAX;
...
double p = 1.0/(baseval*server.lfu_log_factor + 1);
if (r < p) counter++;
通过设置 lfu_log_factor
配置项,可以控制计数器的增加速度,避免 counter
值过快到达255。下图所示,当 lfu_log_factor
等于100时,小于10M 的数据能被区分出来。
一般可以将 lfu_log_factor
取值为 10,已经足够区分1k、1w、10w的数据访问量。
counter的衰减机制:
LFU 策略使用衰减因子配置项 lfu_decay_time
来控制访问次数的衰减。
- LFU 策略会计算当前时间和数据最近一次访问时间的差值,并把这个差值换算成以分钟为单位。
- 然后,LFU 策略再把这个差值除以
lfu_decay_time
值,所得的结果就是数据 counter 要衰减的值。
假设lfu_decay_time
取值为 1,如果数据在 N 分钟内没有被访问,那么它的访问次数就要减 N。
lfu_decay_time
取值更大,那么相应的衰减值会变小,衰减效果也会减弱。如果业务应用中有短时高频访问的数据的话,建议把 lfu_decay_time 值设置为 1,在数据不被访问后,可以 迅速衰减访问次数,从缓存中淘汰出去,避免缓存污染。
总结:LFU 淘汰策略对于扫描式单次数据读取操作时,虽然仍然会频繁写缓存,但可以 避免访问次数高的热点数据因为最近一次访问时间较早而被淘汰,提升了缓存的命中率。