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 列表,记为 keys
keys
中删除一个 key,记为 removed_key
removed_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 列表中
xxxxxxxxxx
from typing import Hashable, Dict, Any, Set
import pprint
import 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()