161 lines
4.1 KiB
Python
161 lines
4.1 KiB
Python
"""
|
||
缓存服务
|
||
提供数据缓存功能（使用内存缓存）
|
||
"""
|
||
import time
|
||
from typing import Optional, Any, Dict
|
||
from app.utils.logger import logger
|
||
|
||
|
||
class CacheService:
|
||
"""内存缓存服务类"""
|
||
|
||
def __init__(self):
|
||
"""初始化内存缓存"""
|
||
self._cache: Dict[str, tuple[Any, float]] = {} # key: (value, expire_time)
|
||
logger.info("内存缓存初始化成功")
|
||
|
||
def get(self, key: str) -> Optional[Any]:
|
||
"""
|
||
获取缓存数据
|
||
|
||
Args:
|
||
key: 缓存键
|
||
|
||
Returns:
|
||
缓存的数据，不存在或过期返回None
|
||
"""
|
||
try:
|
||
if key in self._cache:
|
||
value, expire_time = self._cache[key]
|
||
# 检查是否过期
|
||
if time.time() < expire_time:
|
||
return value
|
||
else:
|
||
# 删除过期数据
|
||
del self._cache[key]
|
||
return None
|
||
except Exception as e:
|
||
logger.error(f"获取缓存失败: {e}")
|
||
return None
|
||
|
||
def set(self, key: str, value: Any, ttl: int = 3600) -> bool:
|
||
"""
|
||
设置缓存数据
|
||
|
||
Args:
|
||
key: 缓存键
|
||
value: 要缓存的数据
|
||
ttl: 过期时间（秒）
|
||
|
||
Returns:
|
||
是否成功
|
||
"""
|
||
try:
|
||
expire_time = time.time() + ttl
|
||
self._cache[key] = (value, expire_time)
|
||
return True
|
||
except Exception as e:
|
||
logger.error(f"设置缓存失败: {e}")
|
||
return False
|
||
|
||
def delete(self, key: str) -> bool:
|
||
"""
|
||
删除缓存
|
||
|
||
Args:
|
||
key: 缓存键
|
||
|
||
Returns:
|
||
是否成功
|
||
"""
|
||
try:
|
||
if key in self._cache:
|
||
del self._cache[key]
|
||
return True
|
||
except Exception as e:
|
||
logger.error(f"删除缓存失败: {e}")
|
||
return False
|
||
|
||
def exists(self, key: str) -> bool:
|
||
"""
|
||
检查缓存是否存在
|
||
|
||
Args:
|
||
key: 缓存键
|
||
|
||
Returns:
|
||
是否存在
|
||
"""
|
||
try:
|
||
if key in self._cache:
|
||
_, expire_time = self._cache[key]
|
||
if time.time() < expire_time:
|
||
return True
|
||
else:
|
||
del self._cache[key]
|
||
return False
|
||
except Exception as e:
|
||
logger.error(f"检查缓存失败: {e}")
|
||
return False
|
||
|
||
def clear_pattern(self, pattern: str) -> int:
|
||
"""
|
||
清除匹配模式的所有缓存
|
||
|
||
Args:
|
||
pattern: 键模式（如 "stock:*"）
|
||
|
||
Returns:
|
||
删除的键数量
|
||
"""
|
||
try:
|
||
# 简单的模式匹配（支持*通配符）
|
||
pattern = pattern.replace('*', '')
|
||
keys_to_delete = [k for k in self._cache.keys() if pattern in k]
|
||
|
||
for key in keys_to_delete:
|
||
del self._cache[key]
|
||
|
||
return len(keys_to_delete)
|
||
except Exception as e:
|
||
logger.error(f"清除缓存失败: {e}")
|
||
return 0
|
||
|
||
def clear_expired(self):
|
||
"""清除所有过期的缓存"""
|
||
try:
|
||
current_time = time.time()
|
||
expired_keys = [
|
||
key for key, (_, expire_time) in self._cache.items()
|
||
if current_time >= expire_time
|
||
]
|
||
|
||
for key in expired_keys:
|
||
del self._cache[key]
|
||
|
||
if expired_keys:
|
||
logger.info(f"清除了{len(expired_keys)}个过期缓存")
|
||
except Exception as e:
|
||
logger.error(f"清除过期缓存失败: {e}")
|
||
|
||
def clear_all(self) -> int:
|
||
"""
|
||
清除所有缓存
|
||
|
||
Returns:
|
||
删除的键数量
|
||
"""
|
||
try:
|
||
count = len(self._cache)
|
||
self._cache.clear()
|
||
logger.info(f"清除了所有缓存，共{count}个")
|
||
return count
|
||
except Exception as e:
|
||
logger.error(f"清除所有缓存失败: {e}")
|
||
return 0
|
||
|
||
|
||
# 创建全局实例
|
||
cache_service = CacheService()
|