HTTP代理怎么用？主流Python框架接入指南

我们青果网络长期服务舆情监测、招投标数据采集这类用Python跑7×24任务的团队，在接入阶段反复看到同一类问题：代码本身没错，是把1分钟存活的短效IP塞进了长连接池，或者把每次请求换IP的隧道当成固定出口来用。

给requests加个proxies参数，为什么接上了还是跑不稳？

因为接入HTTP代理要对上的是三层匹配，proxies参数只解决了最表层的“请求走哪个出口”。

技术团队第一次接代理，通常的判断是：框架支持proxy参数，填上地址就算接完了，之后跑不稳就是代理质量的问题。我们青果网络在协助舆情监测客户排查接入问题时，归因到代理本身的比例并不高，多数卡在下面三层里的某一层没有对上：

这三层里，鉴权层是一次性配置，后两层才是接入的核心。下文每种框架的写法，都围绕“这个框架的连接模型是什么，配哪种代理产品”来展开。

写代码之前，协议、鉴权和提取方式怎么确认？

先在控制台确认三项再动笔：代理协议、鉴权方式、IP提取方式。这三项决定了代码里URL怎么拼、连接怎么管。

代理协议。青果网络的代理IP支持HTTP、HTTPS、SOCKS5三种协议（来源：青果网络官网）。Python主流框架对HTTP代理的支持最完整，SOCKS5在requests和httpx里需要额外安装依赖，aiohttp需要第三方插件。没有特殊要求时，统一走HTTP协议接入，兼容性问题最少。

鉴权方式。青果网络的代理IP支持白名单和账密两种验证方式，白名单可添加256个（来源：青果网络官网）。两者在代码层的区别只有一处：

容器化部署的团队优先用账密。我们青果网络遇到过不少“本地能跑、上线就407”的反馈，基本都是白名单只加了开发机、没加生产节点的出口IP。

IP提取方式。这一项决定了框架的连接模型该怎么配，是三项里最容易被忽略的。各产品的参数与计费以国内代理IP产品页为准：

一条硬边界先说清：全球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不等于接对了，接入验证要跑四个动作，每个动作对应前文的一层匹配。

这四个动作里，节奏验证最容易被跳过。我们青果网络在企业级实践中把它当成接入验收的默认项：参数表上的可用率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不固定，用账密。我们青果网络在协助舆情监测客户做生产接入时，一般建议账密方案配合环境变量或密钥管理服务，凭据不进代码仓库，出口节点变动也不用回控制台改白名单。