In this Article
无论是网页抓取还是网站测试，代理都不可或缺。不过，为了保持稳定访问，您需要使用许多不同的地址，也不可能手动切换每个代理。这正是代理轮换器的用武之地。本文将带您使用 Python 从零构建一个自定义代理轮换器。
为什么构建可靠的代理轮换器很重要
代理轮换器会自动在代理池中切换代理，可为每次请求分配不同的 IP 地址，也可按设定的时间间隔更换。目标服务器不会看到请求都来自同一来源，流量看起来会来自不同的地址和地区，从而降低触发反机器人系统的风险。因此，代理轮换器是网页抓取和多账号运营的关键工具。它有助于维持稳定的身份验证流程、遵守网站的流量规则，并访问地区限定的内容，尤其适合面临多种连接风险的大型项目。
此外，代理轮换器还有一些不那么显眼的优势。由于 IP 会不断变化，网站和追踪器更难分析您的活动，从而提高隐私性和匿名性。真实地址被隐藏后，黑客和专门针对特定设备的诈骗者也更难识别您的身份。
此外，许多网站会根据位置或偏好显示不同结果。使用多个 IP 可以获得更多样、更准确的数据；在不需要个性化内容时，轮换代理还能减少此类内容的干扰。
可靠的代理轮换器：核心组件
除了为每个请求简单更换 IP，代理轮换器还应具备更多功能，才能真正帮上忙而不是制造问题：
- 重试逻辑：如果某个代理无法使用，轮换器会改用其他代理；
- 轮换 User-Agent 请求头，让您的请求看起来不像来自同一来源；
- 处理异常 HTTP 状态码，例如 403（禁止访问）、429（请求过多）、500（内部服务器错误）；
- 暂时禁用持续失败的代理；
- 在请求之间加入短暂延迟，避免触发反机器人系统；
- 日志记录：重试时，您可以查看重试原因以及所用的代理和 User-Agent；
- 抓取前健康检查：开始前测试所有代理，确认其可用；
- 断路器：在一段时间内禁用失效代理；
- 按主机维持粘性会话：在同一会话中让域名始终看到相同的 IP，适用于多账号等需要保持同一会话的场景。
构建代理轮换器时，我们会将这些要点纳入考虑。
使用 Python 构建代理轮换器
我们将在一个文件中创建代理轮换器。示例文件命名为 代理_rotator.py。下文提供全部代码片段及其说明，您可以直接复制粘贴。代码不会硬编码凭据等数据，因此之后无需替换任何内容。
入门
- 从 Python 官方网站 安装或升级 Python。您需要 3.8 或更高版本。
- 运行以下命令安装 requests 库。我们需要它来发起 HTTP 请求。
pip install requests
- 从相应套餐页面获取代理列表。您可以在 DataImpulse 仪表板 中手动复制到 .txt 文件，或直接下载。我们的文件命名为 proxies.txt. 文件内容应如下所示：
http://login:password.host:port
https://login:password.host:port
如有需要，您可以更改代理格式。
导入模块并完成设置
首先，导入我们要使用的模块。
import argparse
import logging
import os
import random
import threading
import time
from dataclasses import dataclass
from typing import Dict, Iterable, List, Optional, Tuple
import requests
from requests import Response
requests 用于发送网络请求，random 用于选择代理和请求头，threading 用于处理多个连接线程。
默认设置
下一部分实现几项关键功能：轮换 User-Agent、遇到错误状态码时切换代理，以及检查代理是否可用。
DEFAULT_USER_AGENTS: List[str] = [
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
"Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:125.0) Gecko/20100101 Firefox/125.0",
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36 Edg/124.0.0.0",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 13_4) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.5 Safari/605.1.15",
"Mozilla/5.0 (iPhone; CPU iPhone OS 16_5 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.5 Mobile/15E148 Safari/604.1",
"Mozilla/5.0 (Linux; Android 13; Pixel 7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Mobile Safari/537.36",
]
DEFAULT_ROTATE_ON_STATUS = {403, 407, 409, 418, 421, 425, 429, 500, 502, 503, 504}
DEFAULT_HEALTHCHECK_URL = "http://api.ipify.org"
如有需要，您可以替换或添加 User-Agent，并使用其他网站进行健康检查。这里我们使用 httpbin.org/ip。其他合适的选择包括 https://httpbin.org/get （显示请求头和 IP）， https://api.ipify.org （或 https://api.ipify.org?format=json 以 JSON 格式返回您的 IP）， https://ifconfig.me 或 https://ifconfig.me/all.json （以 JSON 格式返回请求头和 IP），或 http://ip-api.com/json/ （显示国家、城市、ISP 和 IP）。
代理对象
这一部分定义代理的存储方式。
@dataclass(frozen=True)
class Proxy:
url: str
label: str = ""
def to_requests_proxies(self) -> Dict[str, str]:
return {"http": self.url, "https": self.url}
同样，您无需在此处修改任何内容。
读取代理文件
这一部分从 。TXT 文件中读取代理。
def parse_proxy_file(path: str) -> List[Proxy]:
proxies: List[Proxy] = []
with open(path, "r", encoding="utf-8") as f:
for line in f:
raw = line.strip()
if not raw or raw.startswith("#"):
continue
raw = raw.split(" #", 1)[0].split("\t#", 1)[0].strip()
proxies.append(Proxy(url=raw))
if not proxies:
raise ValueError(f"No proxies found in file: {path}")
return proxies
重试延迟
这段代码实现重试逻辑。每当请求失败，轮换器都会等待更长时间再试，以避免触发限制或被服务拒绝。
def _exp_backoff_sleep(base: float, attempt: int, max_sleep: float) -> None:
sleep_s = min(base * (2 ** attempt) + random.random() * base, max_sleep)
time.sleep(sleep_s)
代理轮换器类
这是轮换器的核心。它选择代理、重试请求并跟踪失败情况；还会决定策略，例如随机选择代理还是依次遍历列表、设置超时和请求头，以及是否为同一网站保持相同代理。它会发送请求、将代理标记为可用或封禁，并执行健康检查。
class ProxyRotator:
def __init__(
self,
proxies: Iterable[Proxy],
strategy: str = "round_robin",
max_retries: int = 3,
timeout: Tuple[float, float] = (10.0, 30.0),
rotate_on_status: Optional[Iterable[int]] = None,
user_agents: Optional[List[str]] = None,
sticky_per_host: bool = True,
circuit_threshold: int = 2,
circuit_cooldown: float = 60.0,
backoff_base: float = 0.5,
backoff_max: float = 5.0,
healthcheck_url: Optional[str] = None,
logger: Optional[logging.Logger] = None,
) -> None:
self.proxies: List[Proxy] = list(proxies)
if not self.proxies:
raise ValueError("ProxyRotator requires at least one proxy")
if strategy not in {"round_robin", "random"}:
raise ValueError("strategy must be 'round_robin' or 'random'")
self.strategy = strategy
self.max_retries = max_retries
self.timeout = timeout
self.rotate_on_status = set(rotate_on_status or DEFAULT_ROTATE_ON_STATUS)
self.user_agents = user_agents or DEFAULT_USER_AGENTS
self.sticky_per_host = sticky_per_host
self.circuit_threshold = circuit_threshold
self.circuit_cooldown = circuit_cooldown
self.failure_counts: Dict[Proxy, int] = {p: 0 for p in self.proxies}
self.disabled_until: Dict[Proxy, float] = {p: 0.0 for p in self.proxies}
self.backoff_base = backoff_base
self.backoff_max = backoff_max
self.healthcheck_url = healthcheck_url
self._local = threading.local()
self._sticky_map: Dict[str, Proxy] = {}
self._idx = 0
self._lock = threading.Lock()
self.log = logger or logging.getLogger("proxy_rotator")
def get(self, url: str, **kwargs) -> Response:
return self.request("GET", url, **kwargs)
def post(self, url: str, **kwargs) -> Response:
return self.request("POST", url, **kwargs)
def request(self, method: str, url: str, **kwargs) -> Response:
session = getattr(self._local, "session", None)
if session is None:
session = requests.Session()
self._local.session = session
headers = kwargs.pop("headers", {}) or {}
headers.setdefault("User-Agent", random.choice(self.user_agents))
timeout = kwargs.pop("timeout", self.timeout)
last_exc: Optional[Exception] = None
last_response: Optional[Response] = None
for attempt in range(self.max_retries + 1):
proxy = self._choose_proxy(url)
if proxy is None:
raise RuntimeError("No healthy proxies currently available")
try:
self.log.debug("Using proxy %s for %s %s", proxy.url, method, url)
response = session.request(
method,
url,
headers=headers,
proxies=proxy.to_requests_proxies(),
timeout=timeout,
**kwargs,
)
last_response = response
if response.status_code in self.rotate_on_status:
self._mark_failure(proxy, f"HTTP {response.status_code}")
self._maybe_unstick_host(url, proxy)
if attempt < self.max_retries:
_exp_backoff_sleep(self.backoff_base, attempt, self.backoff_max)
continue
else:
self._mark_success(proxy)
return response
except requests.RequestException as exc:
last_exc = exc
self._mark_failure(proxy, repr(exc))
self._maybe_unstick_host(url, proxy)
if attempt < self.max_retries:
_exp_backoff_sleep(self.backoff_base, attempt, self.backoff_max)
continue
break
if last_exc:
raise last_exc
assert last_response is not None
return last_response
def health_check(self, url: Optional[str] = None, sample: Optional[int] = None) -> Dict[str, bool]:
check_url = url or self.healthcheck_url or DEFAULT_HEALTHCHECK_URL
to_test = list(self.proxies)
if sample is not None:
to_test = random.sample(to_test, k=min(sample, len(to_test)))
results: Dict[str, bool] = {}
for p in to_test:
try:
r = requests.get(check_url, proxies=p.to_requests_proxies(), timeout=(5, 10))
healthy = r.ok
except Exception:
healthy = False
results[p.url] = healthy
if healthy:
self._mark_success(p)
else:
self._mark_failure(p, "healthcheck")
return results
def _choose_proxy(self, url: str) -> Optional[Proxy]:
host = requests.utils.urlparse(url).hostname or ""
now = time.time()
with self._lock:
if self.sticky_per_host:
sticky = self._sticky_map.get(host)
if sticky and self.disabled_until.get(sticky, 0.0) <= now:
return sticky
enabled = [p for p in self.proxies if self.disabled_until.get(p, 0.0) <= now]
if not enabled:
return None
if self.strategy == "round_robin":
proxy = enabled[self._idx % len(enabled)]
self._idx = (self._idx + 1) % len(enabled)
else:
proxy = random.choice(enabled)
if self.sticky_per_host:
self._sticky_map[host] = proxy
return proxy
def _mark_failure(self, proxy: Proxy, reason: str) -> None:
with self._lock:
cnt = self.failure_counts.get(proxy, 0) + 1
self.failure_counts[proxy] = cnt
if cnt >= self.circuit_threshold:
self.disabled_until[proxy] = time.time() + self.circuit_cooldown
self.log.warning("Disabling proxy %s for %.1fs (reason: %s)", proxy.url, self.circuit_cooldown, reason)
def _mark_success(self, proxy: Proxy) -> None:
with self._lock:
self.failure_counts[proxy] = 0
self.disabled_until[proxy] = 0.0
def _maybe_unstick_host(self, url: str, proxy: Proxy) -> None:
if not self.sticky_per_host:
return
host = requests.utils.urlparse(url).hostname or ""
with self._lock:
if self._sticky_map.get(host) == proxy:
self._sticky_map.pop(host, None)
命令行选项
通过这一部分，您可以在终端中用一行命令运行轮换器。
def build_arg_parser() -> argparse.ArgumentParser:
p = argparse.ArgumentParser(description="HTTP proxy rotator")
p.add_argument("--url", required=False, help="Target URL to request (for quick testing)")
p.add_argument("--method", default="GET", help="HTTP method to use (default: GET)")
p.add_argument(
"--proxies",
help="Path to proxies.txt file (one proxy URL per line). If omitted, reads PROXY_LIST env (comma-separated)",
)
p.add_argument("--strategy", choices=["round_robin", "random"], default="round_robin")
p.add_argument("--retries", type=int, default=3, help="Max retry attempts across proxies")
p.add_argument(
"--timeout",
type=float,
default=15.0,
help="Read timeout in seconds (connect timeout is fixed at 10s in this simple CLI)",
)
p.add_argument(
"--rotate-on",
default=",".join(str(s) for s in sorted(DEFAULT_ROTATE_ON_STATUS)),
help="Comma-separated status codes that trigger rotation",
)
p.add_argument(
"--no-sticky",
action="store_true",
help="Disable sticky-per-host behavior",
)
p.add_argument(
"--healthcheck",
action="store_true",
help="Run a healthcheck across proxies and print a summary before the request",
)
p.add_argument(
"--debug",
action="store_true",
help="Enable verbose logging",
)
return p
def _load_proxies_from_args(args: argparse.Namespace) -> List[Proxy]:
if args.proxies:
return parse_proxy_file(args.proxies)
env = os.getenv("PROXY_LIST", "").strip()
if env:
return [Proxy(url=p.strip()) for p in env.split(",") if p.strip()]
raise SystemExit("No proxies provided. Use --proxies or set PROXY_LIST env.")
def main_cli() -> None:
parser = build_arg_parser()
args = parser.parse_args()
logging.basicConfig(
level=logging.DEBUG if args.debug else logging.INFO,
format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
)
log = logging.getLogger("proxy_rotator")
proxies = _load_proxies_from_args(args)
rotate_on_set = {int(x.strip()) for x in args.rotate_on.split(",") if x.strip()}
rotator = ProxyRotator(
proxies=proxies,
strategy=args.strategy,
max_retries=args.retries,
timeout=(10.0, args.timeout),
rotate_on_status=rotate_on_set,
user_agents=DEFAULT_USER_AGENTS,
sticky_per_host=not args.no_sticky,
circuit_threshold=2,
circuit_cooldown=60.0,
backoff_base=0.5,
backoff_max=5.0,
healthcheck_url=DEFAULT_HEALTHCHECK_URL,
logger=log,
)
if args.healthcheck:
results = rotator.health_check()
total = len(results)
healthy = sum(1 for ok in results.values() if ok)
log.info("Healthcheck: %s/%s proxies OK", healthy, total)
for url, ok in results.items():
log.info(" %-40s %s", url, "OK" if ok else "BAD")
if args.url:
try:
resp = rotator.request(args.method.upper(), args.url)
print(f"Status: {resp.status_code}")
text = resp.text
preview = text[:500].replace("\n", " ")
print(f"Body preview (first 500 chars):\n{preview}")
except Exception as e:
log.error("Request failed: %s", e)
raise
else:
log.info("No --url provided; nothing else to do. Add --url to test a request.")
入口点
这是启动轮换器的入口。它会加载代理、按需检查代理，并发起请求。
if __name__ == "__main__":
main_cli()
运行代理轮换器
您可以在命令行中使用以下命令运行轮换器：
python proxy_rotator.py --proxies ./proxies.txt --url http://api.ipify.org
将 ./proxies.txt 替换为代理文件的实际名称，并将 – – URL 替换为要抓取或测试的网站目标 URL。
此外，还可以添加更多选项：
– – 策略随机 ：轮换器会从代理池中随机选择一个代理
– – 重试 5：可按需要设置不同的重试次数
– – 健康检查：预先测试代理
– – 无粘性：禁用粘性代理。仅在无需保持同一会话时使用
– – 调试：显示详细日志，帮助您排查问题。
注意: 有时，如果您安装的是 Python 3，可能需要使用 “python3 代理_rotator.py –-代理./proxies.txt –-url http://api.ipify.org” 来运行轮换器。
结语
构建代理轮换器并不复杂，却能帮助您充分利用代理池。具备重试逻辑、健康检查和延迟控制功能的可靠轮换器，配合值得信赖的代理，可将网页抓取或多账号运营提升到新水平。若您需要来源可靠、合乎道德规范的 IPs，DataImpulse 随时可以提供帮助。我们可提供来自 195 个地区、超过 9000 万个 IPs，以保障不中断的跨地区连接。联系我们： [email protected]，或点击 “现在就试试” 按钮开始。