LFU(Least Frequently Used)的核心思想是:“如果数据在过去一段时间被访问的次数很少,那么将来被访问的概率也会很低”。
实现思路:
key_to_value)保存 key 到 value 的映射key_to_freq)保存 key 到其使用频率的映射freq_to_keys)保存使用频率到 key 列表的映射,具有相同使用频率的 key 可能有多个min_freq)记录当前的最小使用频率get(key) 执行流程是:
如果 key 不在 key_to_value 中,那么返回异常
否则,通过 key_to_freq 获取 key 的当前使用频率,记为 current_freq
通过 freq_to_keys 获取 current_freq 对应的 key 列表,记为 keys
将 key 从 keys 中移除
如果移除 key 后 keys 为空,那么:
current_freq 从 freq_to_keys 中删除min_freq 等于 current_freq,那么将 min_freq 置为 current_freq + 1将 key 添加到 current_freq + 1 对应的 key 列表中
将 key 的使用频率设置为 current_freq + 1
set(key, value) 的执行流程是:
如果 key 在 key_to_value 中,那么修改 value。然后返回
如果缓存中的条目数达到最大条目数,那么:
min_freq 对应的 key 列表,记为 keyskeys 中删除一个 key,记为 removed_keyremoved_key 从 key_to_value、key_to_freq 中删除removed_key 后,keys 为空,那么将 min_freq 从 freq_to_keys 中删除将 min_freq 设置为 1。(新插入的条目的使用频率为 1。不管之前的最小使用频率是多少,插入条目后的最小使用频率肯定是 1)
将 key/value 对添加到 key_to_value
将 key 对应的使用频率设置为 1
将 key 添加到 1 对应的 key 列表中
xxxxxxxxxxfrom typing import Hashable, Dict, Any, Setimport pprintimport unittest
class LFUCache: def __init__(self, max_items: int) -> None: """ :param max_items: 缓存中的最大条目数 """ self._max_items: int = max_items # 保存 key 到 value 的映射关系 self._key_to_value: Dict[Hashable, Any] = {} # 保存 key 到 frequency 的映射关系 self._key_to_freq: Dict[Hashable, int] = {} # 保存 frequency 到 key 列表的映射关系。因为多个 key 可能具有相同的 frequency self._freq_to_keys: Dict[int, Set[Hashable]] = {} # 当前最小的 frequency self._min_freq: int = 0
def __getitem__(self, key: Hashable) -> Any: """ 获取 key 对应的 Value :param key: key :raise KeyError: 当 key 不存在时,抛出 KeyError :return: value """ # 如果 key 不在缓存中,那么抛出异常 if key not in self._key_to_value: raise KeyError(key)
# 获取 key 对应的 value value: Any = self._key_to_value[key]
# 调整 key 的使用频率,以及按需调整最小的使用频率 current_freq: int = self._key_to_freq[key] new_freq: int = current_freq + 1 keys: Set[Hashable] = self._freq_to_keys[current_freq] keys.remove(key) if len(keys) == 0: self._freq_to_keys.pop(current_freq) if self._min_freq == current_freq: self._min_freq = new_freq if new_freq in self._freq_to_keys: self._freq_to_keys[new_freq].add(key) else: self._freq_to_keys[new_freq] = {key} self._key_to_freq[key] = new_freq
return value
def __setitem__(self, key: Hashable, value: Any) -> None: """ 向缓存中添加新条目。如果 key 已经存在,那么更新 value :param key: key :param value: value """ # 如果 key 已经在缓存中,那么更新其 value if key in self._key_to_value: self._key_to_value[key] = value return
# 如果缓存中的条目数已经达到最大,那么淘汰使用频率最小的 key if len(self._key_to_value) >= self._max_items: # 获取使用频率最小的 key 的列表 keys: Set[Hashable] = self._freq_to_keys[self._min_freq] # 从 key 列表中删除一个 key,它就是被淘汰的 key removed_key: Hashable = keys.pop() # 从 key to value 映射中删除被淘汰的 key self._key_to_value.pop(removed_key) self._key_to_freq.pop(removed_key) # 淘汰 key 后,最小使用频率对应的 key 列表为空,那么进行清理 if len(keys) == 0: self._freq_to_keys.pop(self._min_freq)
self._min_freq = 1
# 添加新条目 self._key_to_value[key] = value self._key_to_freq[key] = 1 if 1 in self._freq_to_keys: self._freq_to_keys[1].add(key) else: self._freq_to_keys[1] = {key}
def debug(self): """ 打印调试信息 """ print(f"min frequency: {self._min_freq}") print("key to value:") pprint.pprint(self._key_to_value) print("key to frequency:") pprint.pprint(self._key_to_freq) print("frequency to keys:") pprint.pprint(self._freq_to_keys) print("-----\n")
def exists(self, key: Hashable) -> bool: """ 判断 key 是否在缓存中 :param key: key :return: 如果 key 在缓存中,那么返回 True,否则返回 False """ return key in self._key_to_value
class TestLFUCache(unittest.TestCase): def testLFUCache(self): cache: LFUCache = LFUCache(3) cache['a'] = 'a' cache.debug() for _ in range(2): self.assertEqual(cache['a'], 'a') cache.debug() cache['b'] = 'b' cache.debug() cache['c'] = 'c' cache.debug()
# 'b' 或 'c' 将被淘汰 cache['d'] = 'd' self.assertTrue((cache.exists('b') and not cache.exists('c')) or (not cache.exists('b') and cache.exists('c'))) cache.debug()
self.assertEqual(cache['d'], 'd') cache.debug()
# 此时,'b' 和 'c' 都将被淘汰 cache['e'] = 'e' self.assertTrue(not cache.exists('b') and not cache.exists('c')) cache.debug()
self.assertTrue(cache['d'], 'd') cache.debug()
self.assertTrue(cache['e'], 'e') cache.debug()
cache['a'] = 'a1' cache.debug() self.assertTrue(cache['a'], 'a1')
if __name__ == "__main__": unittest.main()