由于有缓存图片验证码的需要,所以我找了一些开源的缓存想直接使用,结果要么太老旧,要么太复杂,要么和
fastAPI
整合有问题,所以才想到自己写一个简单的,够用就好。
本文讲述了如何实现一个简单的内存缓存,他可以把过期的键值自动清除,在缓存验证码等场合应该很好用。
它的基本实现思路是:在初始化的时候自动启动一个清理线程,轮询删除过期的键值。
在实现的过程中使用了 lock
,以保证多线程安全。
主要代码如下:
class Cache:def __init__(self, max_size, ttl):"""初始化缓存类:param max_size: 缓存最大容量:param ttl: 每个缓存元素的生存时间(秒)"""self.max_size = max_sizeself.ttl = ttlself.cache = {} # 存储缓存数据self.lock = Lock() # 保证线程安全self._start_cleanup_thread() # 启动自动清理线程def _start_cleanup_thread(self):"""启动后台线程定期清理过期缓存"""def cleanup():while True:time.sleep(1)self._remove_expired_keys()thread = Thread(target=cleanup, daemon=True)thread.start()def _remove_expired_keys(self):"""移除过期的缓存键"""with self.lock:current_time = time.time()expired_keys = [key for key, (_, expire_at) in self.cache.items()if expire_at <= current_time]for key in expired_keys:del self.cache[key]print(f"清理过期键:{key}")def add(self, key, value):"""添加缓存元素:param key: 缓存键:param value: 缓存值:raises CacheFullError: 当缓存已满时抛出"""with self.lock:if key in self.cache:# 如果键已存在,更新值和过期时间self.cache[key] = (value, time.time() + self.ttl)returnif len(self.cache) >= self.max_size:print("缓存已满,无法添加新元素") return Error.FULLself.cache[key] = (value, time.time() + self.ttl)return Error.OKdef get(self, key):"""获取缓存值:param key: 缓存键:return: 缓存值:raises KeyError: 当键不存在或过期时抛出"""with self.lock:if key in self.cache:value, expire_at = self.cache[key]if time.time() < expire_at:return Error.OK, valueelse:del self.cache[key] # 自动清除过期键return Error.EXPIRED,Nonereturn Error.NOT_FOUND,None
以上是主要的逻辑,下面是全部代码下载地址:
- gitee
- github
- gitcode
它已经实际应用在 langchain+llama3+Chroma RAG demo 中,有兴趣您也体验实际应用,并欢迎指正。
- gitee
- github
- gitcode
🪐祝您好运🪐