不少Python项目在接入代理IP时，会把提取代理和发送请求写在同一个函数里。
项目规模小时问题不明显，但并发任务增加后，容易出现以下情况：
多个线程同时调用代理提取API；
同一个失效代理被重复使用；
每次请求都重新获取代理，造成浪费；
代理状态无法统一管理；
请求失败后不知道该换代理还是继续重试。
解决这类问题，可以在业务请求和代理平台之间增加一个简单的代理管理器。
一、代理管理器需要做什么？
一个基础代理管理器至少需要具备以下能力：
从API获取代理；
缓存当前代理；
记录代理获取时间；
判断代理是否需要更换；
请求失败时主动废弃代理；
在多线程环境中避免重复刷新。
携趣短效代理的有效时间可根据产品类型选择，官方资料显示部分短效产品有效时间为30秒至5分钟。因此，本地缓存时间不能超过代理本身的有效时间。
二、实现线程安全的代理管理器
from __future__ import annotationsimport threadingimport timefrom dataclasses import dataclassimport requests@dataclass(frozen=True)class ProxyRecord: address: str fetched_at: floatclass ProxyManager: def __init__( self, api_url: str, local_ttl: int = 40, api_timeout: int = 10, ) -> None: self.api_url = api_url self.local_ttl = local_ttl self.api_timeout = api_timeout self._record: ProxyRecord | None = None self._lock = threading.Lock() def _fetch_from_api(self) -> ProxyRecord: response = requests.get( self.api_url, timeout=self.api_timeout, ) response.raise_for_status() address = response.text.strip().splitlines()[0]if ":" not in address: raise ValueError( f"代理接口返回格式异常: {address!r}" ) return ProxyRecord( address=address, fetched_at=time.monotonic(), ) def _is_expired(self, record: ProxyRecord) -> bool: age = time.monotonic() - record.fetched_at return age >= self.local_ttl def get_proxy(self, force_refresh: bool = False) -> str: record = self._record if ( not force_refresh and record is not None and not self._is_expired(record) ): return record.address with self._lock: record = self._record if ( not force_refresh and record is not None and not self._is_expired(record) ): return record.address new_record = self._fetch_from_api() self._record = new_record return new_record.address def discard(self, address: str) -> None: with self._lock: if ( self._record is not None and self._record.address == address ): self._record = None
三、为什么使用两次过期判断？
代码中第一次判断发生在加锁前：
if record is not None and not self._is_expired(record):
return record.address
大多数请求都会在这里直接返回，不需要竞争锁。
只有代理不存在或者过期时，线程才进入锁。
进入锁后再次判断，是因为等待锁期间，可能已经有其他线程刷新了代理。如果不二次检查，多个线程还是会连续调用代理API。
这种写法通常称为双重检查。
四、结合请求函数使用
from typing import FinalTARGET_TIMEOUT: Final = (5, 15)def proxy_dict(address: str) -> dict[str, str]: url = f"http://{address}" return { "http": url, "https": url, }def get_with_managed_proxy( manager: ProxyManager, url: str, attempts: int = 3,) -> requests.Response: last_error: Exception | None = None for _ in range(attempts): proxy = manager.get_proxy() try: response = requests.get( url, proxies=proxy_dict(proxy), timeout=TARGET_TIMEOUT, ) response.raise_for_status() return response except ( requests.ProxyError, requests.ConnectTimeout, requests.ReadTimeout, requests.ConnectionError, ) as exc: last_error = exc manager.discard(proxy) raise RuntimeError("多次更换代理后仍请求失败") from last_error
五、不要把所有HTTP状态都当作代理失效
假设目标网站返回：
404 Not Found
这通常表示请求地址不存在，不代表代理IP失效。
如果返回：
401 Unauthorized
可能是接口认证错误。
如果返回：
429 Too Many Requests
可能是请求频率过高，也可能与目标网站的访问策略有关。
因此，不建议遇到任何非200状态就立即删除代理。
可以根据业务设置：
PROXY_RELATED_STATUS = {
407, # Proxy Authentication Required
502,
503,
504,
}
对于403、429等状态，需要结合目标网站规则、请求频率和业务日志综合判断。
六、本地TTL应该设置多长？
本地缓存时间应当满足：
本地TTL < 代理IP有效时间
例如代理有效时间约为60秒，本地TTL可以设置为40秒或45秒，预留网络请求和调度时间。
不要把本地TTL恰好设置成60秒，因为：
API返回到实际请求之间存在时间差；
多线程任务可能排队；
单次请求可能持续数秒；
代理有效时间可能存在边界误差。
七、生产环境还可以增加哪些能力？
基础版本可以继续扩展：
缓存多个代理；
记录每个代理成功次数；
记录平均响应时间；
按失败次数暂时隔离代理；
使用Redis共享代理池；
为不同业务分配不同代理；
对API调用增加限速；
采集结构化监控指标。
总结
动态代理池的核心，不是存一批IP地址，而是管理IP的生命周期。
需要处理：
代理什么时候获取；
什么时候复用；
什么时候废弃；
多线程如何避免重复提取；
请求失败如何分类；
如何控制接口调用频率。
本文示例可以作为小型项目的基础代理管理层。携趣官方短效代理支持API提取，实际接口应从产品后台生成，并结合所购买产品的有效时长调整本地TTL。
