import httpx import asyncio import random import logging import psutil import os from typing import List, Optional from app.core.config import settings from tenacity import ( retry, stop_after_attempt, wait_fixed, retry_if_exception_type ) import httpx import asyncio import random import logging import psutil import os from typing import List, Optional from app.core.config import settings from tenacity import ( retry, stop_after_attempt, wait_fixed, retry_if_exception_type ) class ProxyManager: """ 代理池管理类,负责代理的获取、验证、内存监控和动态调整。 """ _instance: Optional["ProxyManager"] = None _initialized: bool = False logger = logging.getLogger(__name__) # 将 logger 作为类属性初始化 def __new__(cls, *args, **kwargs): if not cls._instance: cls._instance = super(ProxyManager, cls).__new__(cls) return cls._instance def __init__(self) -> None: """ 初始化 ProxyManager 实例。 """ if not self._initialized: self.proxies: List[str] = [] # 存储可用代理 self.max_size: int = settings.PROXY_POOL_MAX_SIZE # 代理池最大容量 self.monitor_interval: int = settings.PROXY_POOL_MONITOR_INTERVAL # 监控间隔 self.is_monitoring: bool = False # 是否正在监控 self._initialized = True self.logger.info(f"代理管理器初始化完成,最大容量: {self.max_size}") async def _fetch_proxies(self) -> List[str]: """ 从配置中获取代理列表。 """ logger.info("正在从配置中获取代理...") if settings.PROXY_POOL: return settings.PROXY_POOL return [] @retry( stop=stop_after_attempt(3), wait=wait_fixed(2), retry=retry_if_exception_type(Exception) ) async def _verify_proxy(self, proxy: str) -> bool: """ 验证代理是否可用。 """ try: async with httpx.AsyncClient( proxies={"http://": proxy, "https://": proxy}, timeout=5 ) as client: response = await client.get("http://www.baidu.com") # 使用百度作为测试目标 response.raise_for_status() logger.debug(f"代理 {proxy} 验证成功。") return True except httpx.RequestError as e: logger.warning(f"代理 {proxy} 验证失败: {e}") # 失败的代理不返回,相当于临时禁用 return False except Exception as e: import traceback # 导入 traceback 模块 logger.error(f"验证代理 {proxy} 时发生未知错误: {e}\n{traceback.format_exc()}") # 记录详细堆栈信息 return False async def _refresh_proxies(self) -> None: """ 刷新代理池,获取新代理并验证。 """ new_proxies: List[str] = await self._fetch_proxies() verified_proxies: List[str] = [] for proxy in new_proxies: if await self._verify_proxy(proxy): verified_proxies.append(proxy) self.proxies = verified_proxies[:self.max_size] # 精简容量 logger.info(f"代理池已刷新,当前可用代理数量: {len(self.proxies)}") async def get_proxy(self) -> Optional[str]: """ 从代理池中获取一个随机代理。 """ if not self.proxies: logger.warning("代理池为空,正在尝试刷新...") await self._refresh_proxies() if not self.proxies: logger.error("代理池刷新后仍为空,无法提供代理。") return None return random.choice(self.proxies) def _get_memory_usage(self) -> float: """ 获取当前进程的内存使用率。 """ process = psutil.Process(os.getpid()) memory_info = process.memory_info() # rss (Resident Set Size) 是进程实际使用的物理内存 return memory_info.rss / (1024 * 1024) # 返回MB async def _monitor_memory_and_adjust_pool(self) -> None: """ 监控内存使用,并根据情况动态调整代理池。 """ import traceback # 导入 traceback 模块 while self.is_monitoring: try: current_memory_mb: float = self._get_memory_usage() logger.info(f"当前内存使用: {current_memory_mb:.2f} MB") # 内存压力检测和动态调整策略 if current_memory_mb > settings.PROXY_MEMORY_THRESHOLD_MB: # 内存压力大,紧急精简代理池并缩短监控间隔 if len(self.proxies) > self.max_size * 0.5: # 紧急精简到一半 self.proxies = self.proxies[:int(self.max_size * 0.5)] logger.warning( "内存压力过大 " f"({current_memory_mb:.2f}MB > " f"{settings.PROXY_MEMORY_THRESHOLD_MB}MB)," f"已紧急精简代理池至 {len(self.proxies)} 个。" ) self.monitor_interval = max( 10, settings.PROXY_POOL_MONITOR_INTERVAL // 2 ) # 缩短监控间隔,至少10秒 logger.warning( "内存压力大,代理池监控间隔调整为 " f"{self.monitor_interval} 秒。" ) elif len(self.proxies) > self.max_size: # 如果代理数量超过最大容量 self.proxies = self.proxies[:self.max_size] logger.warning(f"代理数量过多,已精简代理池至 {len(self.proxies)} 个。") self.monitor_interval = settings.PROXY_POOL_MONITOR_INTERVAL # 恢复正常监控间隔 elif len(self.proxies) < self.max_size * 0.8: # 如果代理数量低于最大容量的80% logger.info("代理数量偏少,尝试刷新代理池以补充。") await self._refresh_proxies() self.monitor_interval = settings.PROXY_POOL_MONITOR_INTERVAL # 恢复正常监控间隔 else: self.monitor_interval = ( settings.PROXY_POOL_MONITOR_INTERVAL ) # 正常情况,保持默认监控间隔 except Exception as e: logger.error(f"监控内存或调整代理池时发生异常: {e}\n{traceback.format_exc()}") # 记录详细堆栈信息 finally: await asyncio.sleep(self.monitor_interval) # 确保即使发生异常也能继续等待 async def start_monitoring(self) -> None: """ 启动内存监控和代理池动态调整任务。 """ if not self.is_monitoring: self.is_monitoring = True logger.info("启动代理池内存监控...") asyncio.create_task(self._monitor_memory_and_adjust_pool()) # 首次启动时刷新代理池 await self._refresh_proxies() def stop_monitoring(self) -> None: """ 停止内存监控任务。 """ self.is_monitoring = False logger.info("停止代理池内存监控。") # 创建全局代理管理器实例 proxy_manager = ProxyManager() # 导入 httpx 以便在 _verify_proxy 中使用