mirror of
https://github.com/wahyd4/TrendRadar.git
synced 2026-08-09 05:15:49 +10:00
137 lines
3.2 KiB
Python
137 lines
3.2 KiB
Python
"""
|
|
缓存服务
|
|
|
|
实现TTL缓存机制，提升数据访问性能。
|
|
"""
|
|
|
|
import time
|
|
from typing import Any, Optional
|
|
from threading import Lock
|
|
|
|
|
|
class CacheService:
|
|
"""缓存服务类"""
|
|
|
|
def __init__(self):
|
|
"""初始化缓存服务"""
|
|
self._cache = {}
|
|
self._timestamps = {}
|
|
self._lock = Lock()
|
|
|
|
def get(self, key: str, ttl: int = 900) -> Optional[Any]:
|
|
"""
|
|
获取缓存数据
|
|
|
|
Args:
|
|
key: 缓存键
|
|
ttl: 存活时间（秒），默认15分钟
|
|
|
|
Returns:
|
|
缓存的值，如果不存在或已过期则返回None
|
|
"""
|
|
with self._lock:
|
|
if key in self._cache:
|
|
# 检查是否过期
|
|
if time.time() - self._timestamps[key] < ttl:
|
|
return self._cache[key]
|
|
else:
|
|
# 已过期，删除缓存
|
|
del self._cache[key]
|
|
del self._timestamps[key]
|
|
return None
|
|
|
|
def set(self, key: str, value: Any) -> None:
|
|
"""
|
|
设置缓存数据
|
|
|
|
Args:
|
|
key: 缓存键
|
|
value: 缓存值
|
|
"""
|
|
with self._lock:
|
|
self._cache[key] = value
|
|
self._timestamps[key] = time.time()
|
|
|
|
def delete(self, key: str) -> bool:
|
|
"""
|
|
删除缓存
|
|
|
|
Args:
|
|
key: 缓存键
|
|
|
|
Returns:
|
|
是否成功删除
|
|
"""
|
|
with self._lock:
|
|
if key in self._cache:
|
|
del self._cache[key]
|
|
del self._timestamps[key]
|
|
return True
|
|
return False
|
|
|
|
def clear(self) -> None:
|
|
"""清空所有缓存"""
|
|
with self._lock:
|
|
self._cache.clear()
|
|
self._timestamps.clear()
|
|
|
|
def cleanup_expired(self, ttl: int = 900) -> int:
|
|
"""
|
|
清理过期缓存
|
|
|
|
Args:
|
|
ttl: 存活时间（秒）
|
|
|
|
Returns:
|
|
清理的条目数量
|
|
"""
|
|
with self._lock:
|
|
current_time = time.time()
|
|
expired_keys = [
|
|
key for key, timestamp in self._timestamps.items()
|
|
if current_time - timestamp >= ttl
|
|
]
|
|
|
|
for key in expired_keys:
|
|
del self._cache[key]
|
|
del self._timestamps[key]
|
|
|
|
return len(expired_keys)
|
|
|
|
def get_stats(self) -> dict:
|
|
"""
|
|
获取缓存统计信息
|
|
|
|
Returns:
|
|
统计信息字典
|
|
"""
|
|
with self._lock:
|
|
return {
|
|
"total_entries": len(self._cache),
|
|
"oldest_entry_age": (
|
|
time.time() - min(self._timestamps.values())
|
|
if self._timestamps else 0
|
|
),
|
|
"newest_entry_age": (
|
|
time.time() - max(self._timestamps.values())
|
|
if self._timestamps else 0
|
|
)
|
|
}
|
|
|
|
|
|
# 全局缓存实例
|
|
_global_cache = None
|
|
|
|
|
|
def get_cache() -> CacheService:
|
|
"""
|
|
获取全局缓存实例
|
|
|
|
Returns:
|
|
全局缓存服务实例
|
|
"""
|
|
global _global_cache
|
|
if _global_cache is None:
|
|
_global_cache = CacheService()
|
|
return _global_cache
|