哈希算法题目:设计一个基于哈希的缓存系统,支持TTL(生存时间)
字数 543 2025-11-02 00:38:44
哈希算法题目:设计一个基于哈希的缓存系统,支持TTL(生存时间)
题目描述
设计一个支持TTL(Time-To-Live)的缓存系统,需要实现以下操作:
put(key, value, ttl):将键值对存入缓存,并设置生存时间(毫秒)get(key):如果键存在且未过期,返回值;否则返回-1remove(key):删除键值对cleanup():清理所有过期键值对
系统需要高效处理大量数据,并定期自动清理过期条目。
解题过程
第一步:理解TTL缓存的核心需求
- 快速查找:哈希表提供O(1)的查找效率
- 过期管理:需要跟踪每个键的插入时间和生存时间
- 自动清理:避免内存泄漏,需要定期清理过期数据
第二步:设计数据结构
class TTLCache:
def __init__(self):
self.cache = {} # 主存储:key -> (value, expire_time)
self.access_order = {} # 访问时间记录:key -> last_access_time
cache字典存储键值对和过期时间access_order记录最后访问时间,用于优化清理策略
第三步:实现put方法
import time
def put(self, key, value, ttl):
current_time = time.time() * 1000 # 转换为毫秒
expire_time = current_time + ttl
self.cache[key] = (value, expire_time)
self.access_order[key] = current_time
# 简单清理策略:当缓存达到一定大小时触发清理
if len(self.cache) > len(self.access_order) * 0.7:
self.cleanup()
第四步:实现get方法
def get(self, key):
if key not in self.cache:
return -1
value, expire_time = self.cache[key]
current_time = time.time() * 1000
if current_time > expire_time:
# 键已过期,删除并返回-1
del self.cache[key]
del self.access_order[key]
return -1
# 更新访问时间
self.access_order[key] = current_time
return value
第五步:实现remove方法
def remove(self, key):
if key in self.cache:
del self.cache[key]
del self.access_order[key]
第六步:实现cleanup清理方法
def cleanup(self):
current_time = time.time() * 1000
expired_keys = []
# 找出所有过期键
for key, (_, expire_time) in self.cache.items():
if current_time > expire_time:
expired_keys.append(key)
# 批量删除过期键
for key in expired_keys:
del self.cache[key]
del self.access_order[key]
第七步:优化清理策略
为了避免频繁全量扫描,实现惰性清理+定期清理:
def __init__(self, cleanup_interval=300):
self.cache = {}
self.access_order = {}
self.last_cleanup = time.time() * 1000
self.cleanup_interval = cleanup_interval # 清理间隔(秒)
def get(self, key):
# 检查是否需要触发定期清理
current_time = time.time() * 1000
if current_time - self.last_cleanup > self.cleanup_interval * 1000:
self.cleanup()
self.last_cleanup = current_time
# ... 原有get逻辑
第八步:处理并发访问
在实际应用中需要考虑线程安全:
import threading
class ThreadSafeTTLCache(TTLCache):
def __init__(self):
super().__init__()
self.lock = threading.RLock()
def get(self, key):
with self.lock:
return super().get(key)
def put(self, key, value, ttl):
with self.lock:
super().put(key, value, ttl)
第九步:测试用例
def test_ttl_cache():
cache = TTLCache()
# 测试基本功能
cache.put("key1", "value1", 1000) # 1秒后过期
assert cache.get("key1") == "value1"
# 测试过期
time.sleep(1.1)
assert cache.get("key1") == -1
# 测试清理功能
cache.put("key2", "value2", 500)
time.sleep(0.6)
cache.cleanup()
assert "key2" not in cache.cache
这个TTL缓存系统结合了哈希表的高效查找和定时清理机制,适用于需要数据自动过期的应用场景。