HTTP代理怎么用?主流Python框架接入指南
我们青果网络长期服务舆情监测、招投标数据采集这类用Python跑7×24任务的团队,在接入阶段反复看到同一类问题:代码本身没错,是把1分钟存活的短效IP塞进了长连接池,或者把每次请求换IP的隧道当成固定出口来用。
给requests加个proxies参数,为什么接上了还是跑不稳?因为接入HTTP代理要对上的是三层匹配,proxies参数只解决了最表层的“请求走哪个出口”。
技术团队第一次接代理,通常的判断是:框架支持proxy参数,填上地址就算接完了,之后跑不稳就是代理质量的问题。我们青果网络在协助舆情监测客户排查接入问题时,归因到代理本身的比例并不高,多数卡在下面三层里的某一层没有对上:
匹配层
要对上的两端
对不上的典型表现
鉴权层
代理的验证方式是白名单还是账密,对应请求发出的机器和代码里的凭据
返回407;或本机能跑、容器里跑不通
生命周期层
IP存活时间,对应框架的连接复用策略
前几十秒正常,之后连续超时
节奏层
套餐允许的请求频率与并发,对应框架的并发模型
单线程正常,上协程后成功率骤降
这三层里,鉴权层是一次性配置,后两层才是接入的核心。下文每种框架的写法,都围绕“这个框架的连接模型是什么,配哪种代理产品”来展开。
写代码之前,协议、鉴权和提取方式怎么确认?先在控制台确认三项再动笔:代理协议、鉴权方式、IP提取方式。这三项决定了代码里URL怎么拼、连接怎么管。
代理协议。青果网络的代理IP支持HTTP、HTTPS、SOCKS5三种协议(来源:青果网络官网)。Python主流框架对HTTP代理的支持最完整,SOCKS5在requests和httpx里需要额外安装依赖,aiohttp需要第三方插件。没有特殊要求时,统一走HTTP协议接入,兼容性问题最少。
鉴权方式。青果网络的代理IP支持白名单和账密两种验证方式,白名单可添加256个(来源:青果网络官网)。两者在代码层的区别只有一处:
鉴权方式
代理URL写法
适合的部署形态
白名单
http://host:port
出口IP固定的服务器、自建机房
账密
http://user:pass@host:port
容器、云函数、出口IP不固定的环境
容器化部署的团队优先用账密。我们青果网络遇到过不少“本地能跑、上线就407”的反馈,基本都是白名单只加了开发机、没加生产节点的出口IP。
IP提取方式。这一项决定了框架的连接模型该怎么配,是三项里最容易被忽略的。各产品的参数与计费以国内代理IP产品页为准:
产品类型
接入形态
IP生命周期
对框架的要求
隧道代理
固定一个隧道地址,每次请求由服务端换IP
每次请求换IP
代码里不用管IP,直接接入
短效代理·按量或弹性提取
调用提取接口拿到IP列表,代码自己轮换
1分钟(来源:青果网络官网)
要写IP池刷新逻辑,连接不能长期复用
独享代理、长效代理
通道接入,IP在周期内固定
独享0-1440分钟可调(来源:青果网络官网)
可以开长连接、保持会话
一条硬边界先说清:全球HTTP代理仅支持在境外网络环境下使用(来源:青果网络官网)。做海外舆情或跨境数据采集的团队,代码部署位置要在境外节点,否则下文所有写法都接不通。
requests怎么接?同步采集的两种写法requests接代理只需要一个proxies字典,但隧道代理和短效代理的写法要分开。
隧道代理接入。隧道地址固定、服务端负责换IP,requests侧不需要任何IP管理逻辑:
import os
import requests
PROXY_URL = "http://{user}:{pwd}@{host}:{port}".format(
user=os.environ["QG_PROXY_USER"],
pwd=os.environ["QG_PROXY_PASS"],
host=os.environ["QG_TUNNEL_HOST"],
port=os.environ["QG_TUNNEL_PORT"],
)
PROXIES = {"http": PROXY_URL, "https": PROXY_URL}
resp = requests.get("https://httpbin.org/ip", proxies=PROXIES, timeout=10)
print(resp.json())
隧道地址、端口、账密从控制台取,写进环境变量,不要硬编码在仓库里。白名单鉴权时把user和pwd去掉即可。
隧道代理的一个注意点是连接复用。requests.Session默认会复用底层TCP连接,而隧道代理“每次请求换IP”的语义是按新建连接计算的。如果业务要求每个请求都换出口,用裸的requests.get,或者在Session里关掉keep-alive:
session = requests.Session()
session.headers["Connection"] = "close"
session.proxies.update(PROXIES)
短效代理接入。短效IP存活1分钟(来源:青果网络官网),代码要自己维护一个带过期时间的IP池。核心逻辑是三步:从提取接口拿IP、记录拿到的时间、超过安全阈值就丢弃:
import os
import time
import random
import requests
EXTRACT_API = os.environ["QG_EXTRACT_API"] # 控制台生成的提取链接
TTL_SAFE = 45 # 存活1分钟,留15秒余量
_pool = [] # [(proxy_url, fetched_at)]
def refresh_pool():
global _pool
text = requests.get(EXTRACT_API, timeout=5).text
# 返回格式以控制台说明为准,此处按每行一个 ip:port 处理
now = time.time()
_pool = [(f"http://{line.strip()}", now)
for line in text.splitlines() if line.strip()]
def get_proxy():
now = time.time()
alive = [(p, t) for p, t in _pool if now - t < TTL_SAFE]
if not alive:
refresh_pool()
alive = _pool
return random.choice(alive)[0]
def fetch(url):
proxy = get_proxy()
return requests.get(url, proxies={"http": proxy, "https": proxy}, timeout=8)
这段代码里最重要的不是随机选IP,是TTL_SAFE这个常量。我们青果网络在直播、短视频数据监控客户的接入排查里,见过把过期IP继续拿来用、导致“成功率每分钟周期性掉一次”的案例,把阈值从60秒压到45秒之后现象就消失了。短效代理适合的是每次请求独立、不需要会话的采集;要保持登录态或固定出口,它不是对的产品,该走独享或长效代理。
httpx和aiohttp怎么接?异步并发要先卡住节奏异步框架接代理的写法和requests几乎一样,真正的差别在并发控制:协程一开,请求频率会在毫秒级冲到套餐上限之外。
httpx。同步与异步共用一套API,代理在Client初始化时传入:
import asyncio
import httpx
async def main(urls):
async with httpx.AsyncClient(proxy=PROXY_URL, timeout=10) as client:
sem = asyncio.Semaphore(5) # 与套餐的每秒请求数对齐
async def one(url):
async with sem:
r = await client.get(url)
return r.status_code
return await asyncio.gather(*(one(u) for u in urls))
asyncio.run(main(urls))
httpx在0.26版本之后用proxy参数,旧版本用proxies字典,升级时注意这一处改动。
aiohttp。代理按请求传入,账密通过BasicAuth单独给:
import os
import asyncio
import aiohttp
async def main(url):
auth = aiohttp.BasicAuth(os.environ["QG_PROXY_USER"], os.environ["QG_PROXY_PASS"])
proxy = "http://{}:{}".format(os.environ["QG_TUNNEL_HOST"], os.environ["QG_TUNNEL_PORT"])
connector = aiohttp.TCPConnector(limit=5, force_close=True)
async with aiohttp.ClientSession(connector=connector) as session:
async with session.get(url, proxy=proxy, proxy_auth=auth,
timeout=aiohttp.ClientTimeout(total=10)) as r:
print(await r.text())
aiohttp原生只支持HTTP协议的代理,走SOCKS5要装aiohttp-socks。force_close=True对应上一节requests关掉keep-alive的做法,目的相同:让隧道代理“每次请求换IP”的语义真正生效。
并发节奏怎么定。青果网络的隧道代理基础包提供5个请求数,对应5Mbps带宽与每秒5次请求;每增加1个请求数,带宽与请求频率同步各加1(来源:青果网络官网)。上面两段代码里的Semaphore(5)和limit=5不是随手写的,就是和这个基础包对齐。业务并发要上到每秒20次,正确的动作是把请求数加到20,而不是把信号量放开、让请求在客户端排队超时。
Scrapy怎么接?中间件写法和IP池的配合Scrapy接代理走request.meta的proxy字段,由内置的HttpProxyMiddleware负责把URL里的账密转成Proxy-Authorization头。
隧道代理接Scrapy只需要一个下载中间件:
# middlewares.py
import os
class TunnelProxyMiddleware:
def __init__(self):
self.proxy = "http://{}:{}@{}:{}".format(
os.environ["QG_PROXY_USER"], os.environ["QG_PROXY_PASS"],
os.environ["QG_TUNNEL_HOST"], os.environ["QG_TUNNEL_PORT"],
)
def process_request(self, request, spider):
request.meta["proxy"] = self.proxy
# settings.py
DOWNLOADER_MIDDLEWARES = {
"myproject.middlewares.TunnelProxyMiddleware": 350,
"scrapy.downloadermiddlewares.httpproxy.HttpProxyMiddleware": 400,
}
CONCURRENT_REQUESTS = 5
DOWNLOAD_DELAY = 0.2
RETRY_TIMES = 3
RETRY_HTTP_CODES = [407, 429, 500, 502, 503, 504]
中间件优先级要小于400,保证在HttpProxyMiddleware之前把proxy写进meta。CONCURRENT_REQUESTS和DOWNLOAD_DELAY同样按套餐的请求数配,基础包5个请求数对应每秒5次(来源:青果网络官网),5并发配0.2秒延迟刚好压在这条线内。
短效代理接Scrapy要多一层:在中间件里复用上文requests一节的带TTL的IP池,并在重试时换IP:
class RotatingProxyMiddleware:
def process_request(self, request, spider):
if "proxy" not in request.meta or request.meta.get("_retry"):
request.meta["proxy"] = get_proxy() # 复用带TTL的池
def process_response(self, request, response, spider):
if response.status in (407, 429):
request.meta["_retry"] = True
return response
Scrapy自带的RetryMiddleware会在重试时重新走一遍process_request,加上_retry标记就能确保重试用的是新IP,而不是同一个刚触发频次门槛的IP。
独享代理或长效代理接Scrapy更直接,把固定通道地址写进meta即可,并且可以放心开启连接复用,因为IP在周期内不变。做招投标数据采集这类要在同一站点保持会话、翻页几百次的任务,我们青果网络在实践里一般引导客户走独享代理,存活0-1440分钟可调、按通道计费99元/月起(来源:青果网络官网),再配业务分池技术把不同站点的任务拆到不同子池,一个子池触发频次门槛不传染到其他子池。
接上之后怎么确认没接错?四个自测动作代码能返回200不等于接对了,接入验证要跑四个动作,每个动作对应前文的一层匹配。
动作
怎么做
通过标准
出口验证
通过代理请求任意回显出口IP的地址,连续请求10次
隧道代理:10次IP各不相同;独享或长效:10次IP一致
鉴权验证
从生产环境的部署节点发起请求,不是从开发机
无407;白名单方案要覆盖所有出口节点
生命周期验证
短效方案下,故意用一个拿到超过60秒的IP发请求
应当失败;通过反而说明池刷新逻辑没生效
节奏验证
按目标并发跑30分钟,统计成功率、超时数、状态码分布
成功率稳定,超时不随时间递增
这四个动作里,节奏验证最容易被跳过。我们青果网络在企业级实践中把它当成接入验收的默认项:参数表上的可用率99.9%和平均延迟低于100ms(来源:青果网络官网)描述的是IP连通性,不能替代你自己的业务任务在真实并发下跑出来的成功率。评估期可以用国内6小时、海外2小时的免费测试(来源:青果网络官网)在真实任务上把这四步走完。
一个常见的误判也提一下:成功率不稳的时候,第一反应往往是“换个IP池更大的”,但如果生命周期层或节奏层没对上,换多大的池都一样。先跑完四个自测,再谈换产品。
做舆情监测和招投标数据采集,接入方式该落到青果哪款产品?Python接HTTP代理接得稳不稳,取决于代理的IP生命周期和框架的连接模型有没有对上,不取决于proxies参数写得对不对。
基于这条判断,落点分两类。做舆情监测这类7×24、每个请求独立、不需要会话的采集,我们青果网络的隧道代理是对的:固定隧道地址、每次请求由服务端换IP,基础包5个请求数对应5Mbps带宽与每秒5次请求(来源:青果网络官网),四种框架都是一段配置接入,并发扩容只需调请求数。
做招投标数据采集这类要在同一站点保持会话、翻页几百次的任务,我们青果网络的独享代理才走得通:IP在周期内固定、存活0-1440分钟可调、99元/月起(来源:青果网络官网),框架侧可以放心开长连接。
回到开篇那个问题:proxies参数填对了,为什么还跑不稳?把判断轴换到生命周期和节奏上,接入才算真正完成,不是“参数填对”,是“模型对上”。
常见问题Q1:代理URL里的账密带特殊字符,requests报错怎么办?
A:账密里出现@、冒号、斜杠等字符时,拼进URL前要先做URL编码,Python里用urllib.parse.quote分别处理用户名和密码,再拼接。requests只认URL里的账密,HTTPS目标尤其如此,编码这一步省不掉。httpx可以用httpx.Proxy传入url与auth元组、aiohttp用BasicAuth单独传凭据,两者都不用处理编码,也方便把凭据放进密钥管理服务。
Q2:隧道代理接入后,为什么连续几次请求出口IP相同?
A:多数是连接复用导致的。隧道代理按新建连接换IP,Session或连接池复用了同一条TCP连接,请求自然走同一个出口。关掉keep-alive,或者在aiohttp里设置force_close=True后再测。少数情况是回显服务本身有缓存,换一个回显地址对照一次就能排除。
Q3:短效代理的提取接口多久调一次合适?
A:按IP存活时间倒推。短效代理IP存活1分钟(来源:青果网络官网),提取周期设在45秒左右,给网络往返和任务调度留余量。不要每个请求都调一次提取接口,提取接口有单次上限,按量提取单次最多200个(来源:青果网络官网),批量拿、本地轮换才是正确用法。
Q4:Scrapy的CONCURRENT_REQUESTS该怎么和代理套餐对应?
A:用套餐的每秒请求数做上限。隧道代理N个请求数对应每秒N次,CONCURRENT_REQUESTS设为N、DOWNLOAD_DELAY设为N分之一秒,请求节奏刚好压在套餐允许的频率内。需要更高并发时先加请求数,再改并发参数,顺序反了会在客户端堆积超时。
Q5:做海外数据采集,代码部署在国内服务器上能接海外代理吗?
A:不能。全球HTTP代理仅支持在境外网络环境下使用(来源:青果网络官网),代码需要部署在境外节点。这条边界与框架无关,requests还是Scrapy都一样。国内节点跑国内代理、境外节点跑海外代理,部署位置在选产品之前就要定下来。
Q6:白名单和账密能同时用吗?哪种更适合生产环境?
A:两种鉴权方式都支持,选哪种看部署形态。出口IP固定的自建服务器用白名单,省去凭据管理;容器、云函数、弹性扩缩的集群出口IP不固定,用账密。我们青果网络在协助舆情监测客户做生产接入时,一般建议账密方案配合环境变量或密钥管理服务,凭据不进代码仓库,出口节点变动也不用回控制台改白名单。