估计不少同学都遇见过这样的情况：明明买了代理，一上高并发就大面积超时、可用率忽高忽低、目标站点还是把你封了。
下面按 接入 → 调参 → 调优 → 验证 四个阶段拆开讲，每一步都配可以直接改改就能跑的代码。
先别急着写代码：短效代理 vs 隧道代理，你选对了吗？
大部分配置踩的坑，根源在于一开始就选错了代理类型。两者的差别不在"哪个更好"，而在于 IP 管理的活儿由谁来干。
| 维度 | 短效代理 | 隧道代理 |
|---|---|---|
| IP 提取/轮换 | 你自己在程序里实现 | 云端自动完成，你只配一个入口 |
| 接入复杂度 | 高（要维护 IP 池） | 低（一个固定地址搞定） |
| 可控性 | 强（提取频率、存活时长自己定） | 弱（受服务商套餐限制） |
| 适合场景 | 批量定时任务、高频短周期采集 | 持续并发任务、不想养运维的小团队 |
一句话决策：手上有人力愿意维护 IP 池、追求极致成本和控制力 → 短效代理；想少写代码、快速上线 → 隧道代理。
1. 接入：三步跑通链路
Step 1｜拿到代理凭证
两种鉴权方式，按部署形态选：
- IP 白名单：把你服务器的出口 IP 加进服务商白名单，之后请求免鉴权。适合 IP 固定的单机/固定集群。
- 账密验证：用
用户名:密码鉴权，机器换 IP 也不影响。适合多机、容器化、出口 IP 不固定的环境。
踩坑提醒：白名单最容易翻车的地方是你以为的出口 IP 不是真实出口 IP（比如挂了 NAT、走了公司网关）。先
curl ifconfig.me确认真实出口再加白名单。
Step 2｜配置请求客户端
以 requests 为例，账密代理的最小可用配置：
proxies = {
"http": "http://user:pass@proxy_host:port",
"https": "http://user:pass@proxy_host:port",
}
resp = requests.get("https://httpbin.org/ip", proxies=proxies, timeout=5)
print(resp.json()) # 打印出口 IP，确认走的是代理
SOCKS5 代理要多装一个扩展，并把协议头换掉：
pip install "requests[socks]" # 底层依赖 PySocks
proxies = {
"http": "socks5://user:pass@proxy_host:port",
"https": "socks5://user:pass@proxy_host:port",
}
Step 3｜验证连通性
别拿单个站点测就完事。至少测 3 个不同目标站，一次性看清协议兼容性、鉴权、响应时间：
import requests
def test_proxy(proxy: str, targets: list[str]):
proxies = {
"http": proxy, "https": proxy}
for url in targets:
try:
r = requests.get(url, proxies=proxies, timeout=5)
print(f"[OK] {url} status={r.status_code} {r.elapsed.total_seconds():.2f}s")
except Exception as e:
print(f"[FAIL] {url} {type(e).__name__}: {e}")
test_proxy(
"http://user:pass@proxy_host:port",
[
"https://httpbin.org/ip", # 看出口 IP
"https://www.baidu.com", # 国内可达性
"https://your-target.com", # 真实目标站
],
)
2. 调参（短效代理）：提取频率、并发数、存活时长要"对上"
短效代理高并发的核心，是让这三个参数互相匹配，任何一个不匹配都会拖垮整体。
IP 提取频率：API 一般有每秒提取次数上限。高并发下千万别每次请求前临时提，要提前批量取 IP 存进本地池。经验值：本地池水位维持在 当前并发数的 2~3 倍（10 并发就备 20~30 个 IP）。
并发线程数：不是越高越好。单个 IP 带宽有上限，线程数超过带宽承载后，每个线程的实际吞吐反而下降。从 5~10 起步，逐步上调到目标站点刚好不触发频控的临界点。
IP 存活时长：越短轮换越快，但程序端管理压力越大。高频采集选 1~3 分钟，中频选 5~10 分钟。
下面是一个带本地 IP 池 + 自动补水 + 失败拉黑的可用实现骨架：
import time
import threading
import requests
from queue import Queue, Empty
from concurrent.futures import ThreadPoolExecutor
class ProxyPool:
def __init__(self, api_url: str, min_size: int = 20, fetch_count: int = 50):
self.api_url = api_url
self.min_size = min_size
self.fetch_count = fetch_count
self.pool: Queue = Queue()
self.fail_count: dict[str, int] = {
}
self.blacklist: set[str] = set()
self._lock = threading.Lock()
def _fetch(self):
"""批量提取 IP，跳过黑名单"""
try:
resp = requests.get(f"{self.api_url}&count={self.fetch_count}", timeout=5)
for ip in resp.text.strip().splitlines():
ip = ip.strip()
if ip and ip not in self.blacklist:
self.pool.put(ip)
except Exception as e:
print(f"[pool] 提取失败: {e}")
def get(self) -> str:
"""水位低于下限就自动补水"""
if self.pool.qsize() < self.min_size:
self._fetch()
return self.pool.get(timeout=10)
def report(self, ip: str, ok: bool):
"""成功放回复用；连续失败 3 次拉黑"""
if ok:
self.pool.put(ip)
self.fail_count.pop(ip, None)
return
with self._lock:
self.fail_count[ip] = self.fail_count.get(ip, 0) + 1
if self.fail_count[ip] >= 3:
self.blacklist.add(ip) # 不再放回，避免浪费配额
def crawl(pool: ProxyPool, url: str):
ip = pool.get()
proxies = {
"http": f"http://{ip}", "https": f"http://{ip}"}
try:
r = requests.get(url, proxies=proxies, timeout=5)
pool.report(ip, ok=(r.status_code == 200))
return r.text
except Exception:
pool.report(ip, ok=False)
return None
if __name__ == "__main__":
pool = ProxyPool(api_url="https://api.proxy-provider.com/get?token=xxx",
min_size=20, fetch_count=50)
urls = ["https://your-target.com/page/%d" % i for i in range(1000)]
with ThreadPoolExecutor(max_workers=10) as ex:
results = list(ex.map(lambda u: crawl(pool, u), urls))
参数怎么配成一张表，方便对照：
config = {
"fetch_interval": 30, # 每 30s 批量补一次水
"fetch_count": 50, # 每次提 50 个
"pool_min_size": 20, # 本地池下限（= 并发数 × 2）
"thread_count": 10, # 并发线程
"timeout": 5, # 单请求超时
"retry": 3, # 失败重试
}
3. 调参（隧道代理）：三个参数就够了
因为 IP 轮换全在云端，程序端只需要关心：入口地址、并发上限、超时时间。
import requests
# 隧道代理：地址固定，云端自动切 IP
proxies = {
"http": "http://user:pass@tunnel.example.com:port",
"https": "http://user:pass@tunnel.example.com:port",
}
def fetch(url: str):
# 超时设长一点：隧道切 IP 有额外开销
return requests.get(url, proxies=proxies, timeout=8)
两个关键约束：
- 并发数别超服务商套餐上限。超出的请求会被排队或直接拒绝——你以为是自己代码的问题，其实是套餐限流。默认套餐常见是"每秒 N 个请求 + 固定带宽"，配置前先去控制台确认这个数字。
- 超时要给足。隧道每次切 IP 有开销，超时设得跟短效代理一样短，会把正常请求误杀成失败。
4. 调优：四招显著提效
① 连接复用（性价比最高）
HTTP/1.1 的 Keep-Alive、HTTP/2 的多路复用能省掉大量 TCP 握手。requests 里换用 Session 就能吃到复用红利，实测 QPS 能提 20~40%：
session = requests.Session()
session.proxies = proxies
# 复用底层连接池，别每次都新建 requests.get
for url in urls:
session.get(url, timeout=8)
② 请求头随机化
高并发下所有请求的 User-Agent、Accept-Language 一模一样，等于给目标站点递上"我是脚本"的名片。维护一个头部池随机取：
import random
UA_POOL = [
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 ...",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 ...",
# ... 再来十几个
]
def random_headers() -> dict:
return {
"User-Agent": random.choice(UA_POOL),
"Accept-Language": random.choice(["zh-CN,zh;q=0.9", "en-US,en;q=0.8"]),
}
③ 失败 IP 快速剔除
短效代理场景，一个 IP 连续失败就赶紧踢，别让它继续吃你的请求配额。上面 ProxyPool.report() 里的"连续失败 3 次拉黑"就是最小实现，够用。
④ 分时段调并发
目标站点白天频控严、凌晨松是常态。在总请求量不变的前提下，按站点实际表现动态调并发，可以显著压低失败率——本质是把请求"错峰"到频控更宽松的时段。
5. 用异步再榨一层性能（可选进阶）
线程池到几百并发就会被 GIL 和线程切换拖累。真要冲高并发，asyncio + aiohttp 才是正解，单机上万并发不是问题：
import asyncio
import aiohttp
async def fetch_one(session, url, proxy, sem):
async with sem: # 用信号量控住并发上限，别把目标站打死
try:
async with session.get(
url, proxy=f"http://{proxy}",
timeout=aiohttp.ClientTimeout(total=8),
) as resp:
return await resp.text()
except Exception:
return None
async def main(urls, proxy):
sem = asyncio.Semaphore(50) # 并发上限
async with aiohttp.ClientSession() as session:
tasks = [fetch_one(session, u, proxy, sem) for u in urls]
return await asyncio.gather(*tasks)
# asyncio.run(main(urls, "user:pass@host:port"))
隧道代理用异步尤其省心：一个地址喂给成百上千个协程就行，IP 轮换交给云端。
6. 验证：三轮跑完再上线
配置改完别急着往生产切，按这三轮验证，出问题能提前拦住：
| 阶段 | 目标 | 时长 | 关注指标 |
|---|---|---|---|
| 连通性 | 能跑通 | 5~10 分钟 | 协议兼容、鉴权通过、目标可达 |
| 稳定性 | 跑得稳 | 1~2 小时 | 可用率、响应时间分布 |
| 性能压测 | 跑得快 | 30~60 分钟 | 吞吐量、延迟曲线 |
- 稳定性验证先用目标并发的 50% 跑 1~2 小时。可用率不达标，先分清是代理 IP 的锅还是目标站点的锅——这一步分错方向，后面全白搞。
- 性能压测逐步拉到目标并发，盯延迟曲线：如果响应时间骤增（而非线性上升），说明触到当前配置的并发瓶颈，回到第 2/3 步调参。
强烈建议在免费/测试环境跑完三轮再切生产。别在跑着任务的生产环境直接改代理配置，容易把正在采的活儿一起搞崩。
FAQ
Q：配置里最常见的错误是什么？
超时设得不合理。太短把正常请求误判为失败，太长让慢请求拖住整个线程池。经验值：设为目标站点平均响应时间的 3~5 倍。
Q：短效代理和隧道代理能用同一个爬虫框架吗？
能。区别只在代理配置层。Scrapy / requests 里换个代理中间件配置就行，爬虫逻辑不用动。
Q：本地 IP 池水位设多高？
当前并发线程数的 2~3 倍。10 并发就备 20~30 个，低于下限触发批量补水，避开高峰断供。
Q：怎么排查请求失败原因？
按优先级：先分清是代理超时还是目标站拒绝，再分清是IP 被目标站限制还是代理服务本身异常。记录每次失败的 HTTP 状态码 + 响应体，是排查的地基。
Q：SOCKS5 和 HTTP 代理配置差在哪？
SOCKS5 要在客户端指定协议为 socks5，并确保库支持（Python 里给 requests 装 PySocks）。格式：socks5://user:pass@host:port。
Q：高并发要不要关注带宽？
要。并发数 × 单次响应体积 一旦超过代理带宽上限，就会排队、延迟飙升。采数据量大的站点时，带宽往往比 IP 数量更早成为瓶颈。
代理选型和参数没有标准答案，最终都得拿目标站点压出来。以上代码都是骨架，按你的框架和目标站微调即可。有踩到别的坑，欢迎评论区交流。
