不少采集程序已经用了ThreadPoolExecutor，但运行日志里的出口IP仍然高度重复。问题通常不在线程池，而在代理仍被写死在全局配置里。
线程只负责并发执行。要让IP随请求轮换，还需要一个独立的代理调度层。
为什么线程数增加了，IP却没有轮换？
增加线程只会提高并发量，不会改变请求使用的代理。
下面三种写法都会造成IP无法正常轮换：
- 所有线程共用同一个固定代理。
- 程序启动时随机选择一次代理，后续请求一直复用。
- 多个线程共享同一个
requests.Session，并不断修改其中的代理配置。
第三种情况还可能产生线程安全问题。某个线程刚设置代理A，另一个线程就把配置改成代理B，实际请求使用哪个代理会变得不可控。
多线程IP轮换应遵循一个基本顺序：
领取代理 → 发起请求 → 判断结果 → 归还代理 → 失败时冷却 → 重试时更换代理多线程IP轮换需要哪些组件？
一个可长期运行的实现至少需要线程池、代理池、线程独立会话和失败恢复四部分。
| 组件 | 负责什么 | 为什么需要 |
|---|---|---|
| 线程池 | 并发执行采集任务 | 避免频繁创建和销毁线程 |
| 代理池 | 轮询分配可用代理 | 防止代理被固定在单个线程上 |
| 线程独立会话 | 为每个线程保存连接池 | 避免跨线程修改同一个会话 |
| 单代理并发上限 | 控制同一代理同时处理的请求数 | 减少单个出口瞬时压力 |
| 失败冷却 | 暂停连续失败的代理 | 防止无效代理被反复调用 |
| 有限重试 | 使用其他代理重新请求 | 避免任务因一次连接失败直接结束 |
对于网站采集器，可将不同页面作为独立任务提交。舆情监测和招投标数据采集也可以使用相同结构，只需替换任务URL及数据解析函数。
完整代码如何实现？
下面的代码可以直接运行，代理文件采用一行一个地址的格式。
先安装依赖：
pip install requests创建proxies.txt：
http://username:password@192.0.2.10:8000
http://username:password@192.0.2.11:8000
http://192.0.2.12:8000保存下面代码为ip_rotation.py：
from __future__ import annotations
import argparse
import json
import random
import re
import threading
import time
from collections import deque
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass
from pathlib import Path
from typing import Any
from urllib.parse import urlsplit
import requests
from requests.adapters import HTTPAdapter
@dataclass
class ProxyNode:
"""代理节点及其运行状态。"""
endpoint: str
in_flight: int = 0
fail_streak: int = 0
cooldown_until: float = 0.0
@property
def requests_config(self) -> dict[str, str]:
return {
"http": self.endpoint,
"https": self.endpoint,
}
class ProxyPool:
"""线程安全的轮询代理池。"""
def __init__(
self,
proxies: list[str],
max_in_flight: int = 1,
failure_threshold: int = 2,
cooldown_seconds: float = 20.0,
) -> None:
if not proxies:
raise ValueError("代理列表不能为空")
self._nodes = deque(ProxyNode(endpoint=p) for p in proxies)
self._max_in_flight = max_in_flight
self._failure_threshold = failure_threshold
self._cooldown_seconds = cooldown_seconds
self._condition = threading.Condition()
def acquire(self, timeout: float = 15.0) -> ProxyNode:
"""领取一个未超并发上限且不在冷却期的代理。"""
deadline = time.monotonic() + timeout
with self._condition:
while True:
now = time.monotonic()
for _ in range(len(self._nodes)):
node = self._nodes[0]
self._nodes.rotate(-1)
available = (
node.in_flight < self._max_in_flight
and node.cooldown_until <= now
)
if available:
node.in_flight += 1
return node
remaining = deadline - now
if remaining <= 0:
raise TimeoutError("等待可用代理超时")
ready_times = [
max(0.0, node.cooldown_until - now)
for node in self._nodes
if node.in_flight < self._max_in_flight
]
next_check = min(ready_times, default=0.2)
wait_seconds = min(
remaining,
max(0.05, next_check if next_check > 0 else 0.2),
)
self._condition.wait(timeout=wait_seconds)
def release(self, node: ProxyNode, success: bool) -> None:
"""归还代理，并根据请求结果更新失败状态。"""
with self._condition:
node.in_flight = max(0, node.in_flight - 1)
if success:
node.fail_streak = 0
else:
node.fail_streak += 1
if node.fail_streak >= self._failure_threshold:
node.cooldown_until = (
time.monotonic() + self._cooldown_seconds
)
node.fail_streak = 0
self._condition.notify_all()
_thread_local = threading.local()
def get_session(pool_size: int) -> requests.Session:
"""每个线程创建并复用自己的Session。"""
if not hasattr(_thread_local, "session"):
session = requests.Session()
adapter = HTTPAdapter(
pool_connections=pool_size,
pool_maxsize=pool_size,
max_retries=0,
)
session.mount("http://", adapter)
session.mount("https://", adapter)
session.headers.update(
{
"User-Agent": "Authorized-Data-Collector/1.0",
"Accept": "application/json,text/html;q=0.9,*/*;q=0.8",
}
)
_thread_local.session = session
return _thread_local.session
def normalize_proxy(raw_proxy: str) -> str:
"""补全协议并检查代理格式。"""
proxy = raw_proxy.strip()
if not proxy:
raise ValueError("代理地址不能为空")
if "://" not in proxy:
proxy = f"http://{proxy}"
parsed = urlsplit(proxy)
if not parsed.hostname or not parsed.port:
raise ValueError(f"代理格式无效：{raw_proxy}")
return proxy
def mask_proxy(proxy: str) -> str:
"""输出日志前遮盖代理认证信息。"""
return re.sub(r"//[^/@]+@", "//***:***@", proxy)
def load_proxies(file_path: str) -> list[str]:
"""从文本文件读取代理，自动忽略空行和注释。"""
path = Path(file_path)
if not path.is_file():
raise FileNotFoundError(f"代理文件不存在：{file_path}")
proxies: list[str] = []
for line_number, line in enumerate(
path.read_text(encoding="utf-8").splitlines(),
start=1,
):
value = line.strip()
if not value or value.startswith("#"):
continue
try:
proxies.append(normalize_proxy(value))
except ValueError as exc:
raise ValueError(
f"代理文件第{line_number}行格式错误：{exc}"
) from exc
if not proxies:
raise ValueError("代理文件中没有可用地址")
return proxies
def extract_preview(response: requests.Response) -> str:
"""提取少量响应内容，用于确认任务结果。"""
content_type = response.headers.get("Content-Type", "").lower()
if "application/json" in content_type:
try:
data: Any = response.json()
return json.dumps(data, ensure_ascii=False)[:200]
except ValueError:
pass
return response.text.replace("\n", " ")[:200]
def fetch_with_rotation(
task_id: int,
target_url: str,
proxy_pool: ProxyPool,
request_timeout: float,
max_attempts: int,
pool_size: int,
) -> dict[str, Any]:
"""执行一个任务，失败时重新领取代理。"""
session = get_session(pool_size)
last_error = ""
for attempt in range(1, max_attempts + 1):
try:
proxy_node = proxy_pool.acquire(timeout=request_timeout)
except TimeoutError as exc:
last_error = str(exc)
time.sleep(min(0.5 * attempt, 2.0))
continue
success = False
started_at = time.perf_counter()
try:
response = session.get(
target_url,
proxies=proxy_node.requests_config,
timeout=request_timeout,
allow_redirects=True,
)
# 403、429及5xx等状态进入更换代理后的重试流程。
response.raise_for_status()
success = True
return {
"task_id": task_id,
"success": True,
"attempt": attempt,
"status_code": response.status_code,
"proxy": mask_proxy(proxy_node.endpoint),
"elapsed_seconds": round(
time.perf_counter() - started_at,
3,
),
"preview": extract_preview(response),
}
except requests.RequestException as exc:
last_error = f"{type(exc).__name__}: {exc}"
finally:
proxy_pool.release(proxy_node, success=success)
if attempt < max_attempts:
# 小幅随机退避，避免失败任务在同一时刻再次发起。
delay = min(0.4 * (2 ** (attempt - 1)), 3.0)
time.sleep(delay + random.uniform(0.05, 0.25))
return {
"task_id": task_id,
"success": False,
"attempt": max_attempts,
"status_code": None,
"proxy": None,
"elapsed_seconds": None,
"error": last_error,
}
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(
description="Python多线程IP轮换示例"
)
parser.add_argument(
"--target",
required=True,
help="已获授权的目标URL",
)
parser.add_argument(
"--proxy-file",
default="proxies.txt",
help="代理列表文件，默认proxies.txt",
)
parser.add_argument(
"--tasks",
type=int,
default=20,
help="请求任务数，默认20",
)
parser.add_argument(
"--workers",
type=int,
default=8,
help="线程数，默认8",
)
parser.add_argument(
"--per-proxy-concurrency",
type=int,
default=1,
help="单个代理允许的同时请求数，默认1",
)
parser.add_argument(
"--timeout",
type=float,
default=10.0,
help="单次请求及代理等待超时秒数，默认10",
)
parser.add_argument(
"--attempts",
type=int,
default=3,
help="每个任务最多尝试次数，默认3",
)
parser.add_argument(
"--cooldown",
type=float,
default=20.0,
help="连续失败代理的冷却秒数，默认20",
)
return parser.parse_args()
def main() -> None:
args = parse_args()
if args.tasks < 1 or args.workers < 1:
raise ValueError("tasks和workers必须大于0")
if args.per_proxy_concurrency < 1:
raise ValueError("per-proxy-concurrency必须大于0")
proxies = load_proxies(args.proxy_file)
proxy_pool = ProxyPool(
proxies=proxies,
max_in_flight=args.per_proxy_concurrency,
failure_threshold=2,
cooldown_seconds=args.cooldown,
)
results: list[dict[str, Any]] = []
with ThreadPoolExecutor(
max_workers=args.workers,
thread_name_prefix="collector",
) as executor:
futures = [
executor.submit(
fetch_with_rotation,
task_id,
args.target,
proxy_pool,
args.timeout,
args.attempts,
args.workers,
)
for task_id in range(1, args.tasks + 1)
]
for future in as_completed(futures):
result = future.result()
results.append(result)
print(json.dumps(result, ensure_ascii=False))
results.sort(key=lambda item: item["task_id"])
success_count = sum(
1 for item in results if item["success"]
)
summary = {
"tasks": len(results),
"success": success_count,
"failed": len(results) - success_count,
"success_rate": (
f"{success_count / len(results):.2%}"
if results
else "0.00%"
),
}
print("\n任务汇总：")
print(json.dumps(summary, ensure_ascii=False, indent=2))
if __name__ == "__main__":
main()如何运行并验证轮换结果？
先使用少量任务检查代理格式和目标响应，再逐步增加线程数。
运行命令：
python ip_rotation.py \
--target "https://your-authorized-endpoint.example/check" \
--proxy-file proxies.txt \
--tasks 20 \
--workers 8 \
--per-proxy-concurrency 1 \
--attempts 3 \
--timeout 10每完成一个任务，程序会输出一条JSON记录：
{
"task_id": 3,
"success": true,
"attempt": 1,
"status_code": 200,
"proxy": "http://***:***@192.0.2.11:8000",
"elapsed_seconds": 0.842,
"preview": "{\"origin\":\"203.0.113.25\"}"
}验证时重点看三个字段：
proxy是否在不同任务间轮换。attempt是否在请求失败后增加。success_rate是否因线程数提高而明显下降。
如果线程数增加后成功率持续下降，通常说明单代理并发过高、代理池容量不足，或目标端存在更严格的访问频率控制。此时不应继续盲目增加线程。
代码中的轮换机制为什么可靠？
可靠性来自代理调度与线程执行相互独立，而不是简单随机选择。
轮询代替无约束随机
随机选择可能连续多次抽中同一代理。代码使用deque.rotate轮询代理，让可用节点按顺序参与请求。
单代理并发受到控制
in_flight记录代理当前承担的任务数。默认每个代理同时只处理一个请求，适合先验证可用性。确认代理支持更高并发后，再调整--per-proxy-concurrency。
连续失败进入冷却
某个代理连续失败两次后，会暂停20秒。其他线程不会立即再次领取该节点，减少无效重试。
每个线程拥有独立Session
threading.local让每个线程单独维护requests.Session。连接可以复用，但代理配置不会在不同线程之间相互覆盖。
生产环境还需要补充哪些保护？
正式任务不能只看请求是否返回200，还要同时控制业务正确性和访问节奏。
建议补充以下检查：
- 业务结果校验：部分响应虽然是200，但正文可能是登录页、提示页或空数据，需要按字段判断是否有效。
- 分级失败处理：连接超时可更换代理，参数错误应直接终止，429则应延长等待时间。
- 代理质量统计：记录每个代理的成功率、平均延迟和连续失败次数，定期移除低质量节点。
- 任务级去重：重试可能导致同一页面被处理多次，写入数据前应使用URL或业务主键去重。
- 持久化队列：舆情监测等长期任务可将待处理URL放入消息队列，避免程序退出后丢失进度。
- 授权与频率约束：只处理已获授权的数据，并遵守目标端公开规则及合理请求频率。
网站采集器关注页面吞吐，舆情监测关注持续运行，招投标数据采集关注去重和完整性。三类场景都不应把线程数当作唯一性能指标，真正需要监控的是有效请求率、单代理压力和失败恢复能力。
FAQ
Q：一个线程是否应该固定绑定一个IP？
通常不建议长期绑定。固定关系容易让慢代理拖住整个线程，也不利于故障切换。更合适的方式是每个请求领取代理，完成后归还；只有业务明确要求连续会话时，才在一组关联请求内保持同一代理。
Q：随机选择代理和轮询代理有什么区别？
随机选择实现简单，但小规模代理池中可能连续命中同一节点。轮询能让请求分配更均匀。生产环境还可进一步结合成功率和延迟，按代理质量设置领取优先级。
Q：代理数量少于线程数量会发生什么？
代码会根据per-proxy-concurrency限制实际并发。没有可用代理的线程会等待，而不是重复占用同一节点。这样会降低瞬时吞吐，但更容易保持有效请求率。
Q：为什么不让所有线程共用一个Session？
requests.Session适合复用连接，但不适合由多个线程同时修改代理、Cookie和请求头。线程独立Session可以减少状态相互覆盖，同时保留线程内部的连接复用能力。
Q：多久轮换一次IP更合理？
轮换周期取决于任务类型。独立页面可按请求轮换，连续分页或需要保持状态的流程可按任务组轮换。不能只按固定秒数决定，还要结合会话连续性、代理并发能力和失败率。
Q：出现403或429时是否只要更换IP？
不一定。两类状态也可能与权限、请求频率、必要参数或目标端规则有关。代码会将其纳入重试流程，但生产环境还应识别响应内容，并对429采用更长的等待时间。
// 青果网络企业级API示例
const config = {
auth: "QGE_Enterprise_Key",
region: "CN_ALL"
};
async function getProxy() {
const res = await qgClient.fetch(config);
console.log("Stable Connection.");
return res.ip;
}
