本文目录导读：
我来帮你设计Python数据API的数据预取方案,从简单到复杂的几个层次：
基础预取设计
简单的同步预取
import asyncio
from functools import lru_cache
from datetime import datetime, timedelta
class BasePrefetcher:
"""基础预取器"""
def __init__(self, ttl_seconds=300):
self.cache = {}
self.ttl = timedelta(seconds=ttl_seconds)
@lru_cache(maxsize=128)
async def fetch_data(self, key):
"""实际获取数据的逻辑"""
# 这里是你的API调用逻辑
return {"data": f"value_{key}"}
async def get_data(self, key):
"""获取数据，带缓存"""
if key in self.cache:
data, timestamp = self.cache[key]
if datetime.now() - timestamp < self.ttl:
return data
data = await self.fetch_data(key)
self.cache[key] = (data, datetime.now())
return data
def prefetch(self, keys):
"""预取多个key的数据"""
for key in keys:
asyncio.create_task(self.get_data(key))
带优先级队列的预取
import heapq
from typing import Any, Dict, List
import asyncio
from dataclasses import dataclass, field
@dataclass(order=True)
class PrefetchTask:
priority: int
timestamp: float = field(compare=False)
key: str = field(compare=False)
class PriorityPrefetcher(BasePrefetcher):
"""带优先级的预取器"""
def __init__(self, max_workers=5):
super().__init__()
self.queue = []
self.workers = max_workers
self.semaphore = asyncio.Semaphore(max_workers)
self.running_tasks = set()
async def prefetch_with_priority(self, keys, priority=0):
"""按优先级预取"""
tasks = []
for key in keys:
task = PrefetchTask(
priority=priority,
timestamp=asyncio.get_event_loop().time(),
key=key
)
heapq.heappush(self.queue, task)
tasks.append(self._process_queue())
return await asyncio.gather(*tasks)
async def _process_queue(self):
"""处理队列中的任务"""
while self.queue:
async with self.semaphore:
task = heapq.heappop(self.queue)
if task.key not in self.cache:
await self.get_data(task.key)
智能预测预取
from collections import defaultdict, Counter
import numpy as np
from sklearn.linear_model import LinearRegression
class SmartPrefetcher:
"""智能预测预取器"""
def __init__(self):
self.history = defaultdict(list)
self.patterns = {}
self.access_count = Counter()
self.co_occurrence = defaultdict(Counter)
def record_access(self, key, timestamp):
"""记录数据访问历史"""
self.access_count[key] += 1
self.history[key].append(timestamp)
# 记录共同访问模式
recent_keys = [k for k, ts in self.history.items()
if abs(ts[-1] - timestamp) < 60] # 60秒内的访问
for k in recent_keys:
if k != key:
self.co_occurrence[key][k] += 1
def predict_next_access(self, key):
"""预测下一个可能被访问的key"""
if key not in self.co_occurrence:
return []
# 基于共同出现频率预取
co_occur = self.co_occurrence[key]
threshold = sum(co_occur.values()) / len(co_occur) * 0.5
predicted = [k for k, v in co_occur.most_common(5)
if v > threshold]
return predicted
async def smart_prefetch(self, current_key):
"""智能预取"""
predicted_keys = self.predict_next_access(current_key)
async def prefetch_task():
for key in predicted_keys:
if key not in self.cache:
await self.fetch_data(key)
asyncio.create_task(prefetch_task())
完善的预取框架
from enum import Enum
import aioredis
import pickle
import logging
class CacheLevel(Enum):
MEMORY = 1
REDIS = 2
class ComprehensivePrefetcher:
"""完善的预取框架"""
def __init__(self, redis_url=None):
self.memory_cache = {}
self.redis_client = None
self.logger = logging.getLogger(__name__)
if redis_url:
self.redis_client = aioredis.from_url(redis_url)
# 配置参数
self.config = {
'memory_ttl': 300, # 内存缓存5分钟
'redis_ttl': 3600, # Redis缓存1小时
'prefetch_window': 60, # 预取时间窗口
'max_prefetch': 10, # 每次最大预取数量
'min_confidence': 0.7 # 最低预取置信度
}
async def get_data(self, key, cache_level=CacheLevel.MEMORY):
"""多级缓存获取数据"""
# 尝试内存缓存
if cache_level == CacheLevel.MEMORY:
data = self._get_from_memory(key)
if data:
return data
# 尝试Redis缓存
if self.redis_client and cache_level == CacheLevel.REDIS:
data = await self._get_from_redis(key)
if data:
self._set_to_memory(key, data)
return data
# 从数据源获取
data = await self._fetch_from_source(key)
self._set_to_memory(key, data)
if self.redis_client:
await self._set_to_redis(key, data)
return data
def _get_from_memory(self, key):
"""从内存缓存获取"""
if key in self.memory_cache:
data, timestamp = self.memory_cache[key]
if (datetime.now() - timestamp).seconds < self.config['memory_ttl']:
return data
else:
del self.memory_cache[key]
return None
def _set_to_memory(self, key, data):
"""设置内存缓存"""
self.memory_cache[key] = (data, datetime.now())
# 清理过期缓存
self._clean_expired_cache()
async def _get_from_redis(self, key):
"""从Redis获取"""
if self.redis_client:
data = await self.redis_client.get(key)
if data:
return pickle.loads(data)
return None
async def _set_to_redis(self, key, data):
"""设置Redis缓存"""
if self.redis_client:
await self.redis_client.setex(
key,
self.config['redis_ttl'],
pickle.dumps(data)
)
async def _fetch_from_source(self, key):
"""从数据源获取"""
# 实际的数据获取逻辑
self.logger.info(f"Fetching data for key: {key}")
return await self.api_call(key)
def _clean_expired_cache(self):
"""清理过期缓存"""
now = datetime.now()
expired_keys = [
k for k, (v, t) in self.memory_cache.items()
if (now - t).seconds > self.config['memory_ttl']
]
for k in expired_keys:
del self.memory_cache[k]
async def prefetch_batch(self, keys, priority=0):
"""批量预取"""
prefetched = 0
for key in keys[:self.config['max_prefetch']]:
if key not in self.memory_cache:
async def prefetch():
try:
data = await self._fetch_from_source(key)
self._set_to_memory(key, data)
return True
except Exception as e:
self.logger.error(f"Prefetch failed for {key}: {e}")
return False
asyncio.create_task(prefetch())
prefetched += 1
self.logger.info(f"Prefetched {prefetched} keys")
return prefetched
# 使用示例
async def main():
# 创建预取器
prefetcher = ComprehensivePrefetcher(redis_url="redis://localhost")
# 预取常见数据
common_keys = ["user_123", "user_456", "product_789"]
await prefetcher.prefetch_batch(common_keys, priority=1)
# 获取数据
user_data = await prefetcher.get_data("user_123")
print(user_data)
# 智能预取
prefetcher.record_access("user_123", datetime.now())
predicted = prefetcher.predict_next_access("user_123")
await prefetcher.prefetch_batch(predicted)
# 运行
if __name__ == "__main__":
asyncio.run(main())
最佳实践建议
配置管理
class PrefetchConfig:
"""预取配置类"""
def __init__(self):
self.cache_ttl = 300
self.batch_size = 10
self.concurrent_limit = 5
self.retry_count = 3
self.timeout = 5.0
@classmethod
def from_env(cls):
"""从环境变量加载配置"""
import os
config = cls()
config.cache_ttl = int(os.getenv('CACHE_TTL', 300))
config.batch_size = int(os.getenv('PREFETCH_BATCH', 10))
return config
监控和告警
class PrefetchMonitor:
"""预取监控"""
def __init__(self):
self.stats = {
'prefetched': 0,
'cache_hits': 0,
'cache_misses': 0,
'prefetch_hits': 0, # 预取的数据被使用
'failed_prefetches': 0
}
def record_prefetch_hit(self):
"""记录预取命中"""
self.stats['prefetch_hits'] += 1
def get_hit_rate(self):
"""获取缓存命中率"""
total = self.stats['cache_hits'] + self.stats['cache_misses']
if total == 0:
return 0
return self.stats['cache_hits'] / total
选择哪个方案取决于你的具体需求：
- 小规模应用：使用基础预取
- 高并发场景：使用优先级队列
- 需要预测优化：使用智能预测
- 企业级应用：使用完善框架
预取不是越多越好，需要根据实际访问模式和数据特点来优化。
