跳到主要内容
NextProxyNextProxy文档
EN控制台

Python

requests、httpx、aiohttp 和 Scrapy 的接入方式，含连接复用、轮换控制和重试策略。

Python 生态里代理配置的一个共同陷阱：HTTP 和 HTTPS 都要指到同一个 http:// 代理 URL。写成 https:// 会失败，因为网关没有 TLS 入站。

requests

python
import requests

PROXY = "http://USERNAME-country-US-session-job1-time-30:PASSWORD@GATEWAY_HOST:58971"
PROXIES = {"http": PROXY, "https": PROXY}

response = requests.get(
    "https://api.ipify.org?format=json",
    proxies=PROXIES,
    timeout=30,
)
response.raise_for_status()
print(response.json())

复用 Session

同一个会话内的多个请求应该共用 Session，省掉重复的 TCP + TLS 握手：

python
import requests

PROXY = "http://USERNAME-country-US-session-job1-time-30:PASSWORD@GATEWAY_HOST:58971"

with requests.Session() as session:
    session.proxies = {"http": PROXY, "https": PROXY}
    session.headers["User-Agent"] = "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"

    for page in range(1, 6):
        response = session.get(f"https://example.com/list?page={page}", timeout=30)
        print(page, response.status_code)

注意这样所有请求都走同一条连接、同一个出口 IP。想换 IP 见下面。

轮换 IP

不带 session 参数，并且让每次请求用新连接：

python
import requests

PROXY = "http://USERNAME-country-US:PASSWORD@GATEWAY_HOST:58971"
PROXIES = {"http": PROXY, "https": PROXY}

for _ in range(5):
    # 每次新建 Session，用完就关，连接池不复用
    with requests.Session() as session:
        print(session.get("https://api.ipify.org", proxies=PROXIES, timeout=30).text)

并发 + 每任务独立会话

python
import concurrent.futures

import requests

GATEWAY = "GATEWAY_HOST:58971"
USER = "USERNAME"
PASSWORD = "PASSWORD"


def proxy_for(session_id: str, country: str = "US", minutes: int = 30) -> str:
    user = f"{USER}-country-{country}-session-{session_id}-time-{minutes}"
    return f"http://{user}:{PASSWORD}@{GATEWAY}"


def fetch(task_id: int) -> tuple[int, str]:
    proxy = proxy_for(f"task.{task_id}")
    with requests.Session() as session:
        session.proxies = {"http": proxy, "https": proxy}
        return task_id, session.get("https://api.ipify.org", timeout=30).text


with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool:
    for task_id, ip in pool.map(fetch, range(8)):
        print(task_id, ip)

八个任务各有独立会话，各自锁定一个出口 IP 30 分钟。

SOCKS5

需要额外装依赖：

shell
pip install 'requests[socks]'
python
import requests

# socks5h 让代理解析域名
PROXY = "socks5h://USERNAME-country-US:PASSWORD@GATEWAY_HOST:58971"

print(requests.get(
    "https://api.ipify.org",
    proxies={"http": PROXY, "https": PROXY},
    timeout=30,
).text)

带重试的封装

网关的 502、503、504 可有限退避重试；400、402、403、407 应先修正参数、额度或权限。HTTPS CONNECT 的握手失败可能作为 ProxyError 抛出，而非普通响应状态。

python
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry


def build_session(proxy: str) -> requests.Session:
    session = requests.Session()
    session.proxies = {"http": proxy, "https": proxy}

    retry = Retry(
        total=3,
        backoff_factor=1.5,
        # 仅对可安全重放的请求有限重试；代理握手异常需单独处理
        status_forcelist=(429, 502, 503, 504),
        allowed_methods=frozenset(["GET", "HEAD", "POST"]),
    )
    adapter = HTTPAdapter(max_retries=retry, pool_maxsize=32)
    session.mount("http://", adapter)
    session.mount("https://", adapter)
    return session


PROXY = "http://USERNAME-country-US:PASSWORD@GATEWAY_HOST:58971"
with build_session(PROXY) as session:
    print(session.get("https://api.ipify.org", timeout=30).text)

httpx

python
import httpx

PROXY = "http://USERNAME-country-US-session-job1-time-30:PASSWORD@GATEWAY_HOST:58971"

with httpx.Client(proxy=PROXY, timeout=30.0) as client:
    print(client.get("https://api.ipify.org").text)

异步：

python
import asyncio

import httpx

PROXY = "http://USERNAME-country-US:PASSWORD@GATEWAY_HOST:58971"


async def main() -> None:
    limits = httpx.Limits(max_connections=20, max_keepalive_connections=10)
    async with httpx.AsyncClient(proxy=PROXY, limits=limits, timeout=30.0) as client:
        urls = [f"https://example.com/item/{i}" for i in range(20)]
        responses = await asyncio.gather(*(client.get(url) for url in urls))
        for response in responses:
            print(response.status_code)


asyncio.run(main())

aiohttp

aiohttp 的代理配置和别人不一样——凭据要单独传，不能塞在 URL 里：

python
import asyncio

import aiohttp

GATEWAY = "http://GATEWAY_HOST:58971"
AUTH = aiohttp.BasicAuth(
    "USERNAME-country-US-session-job1-time-30",
    "PASSWORD",
)


async def main() -> None:
    timeout = aiohttp.ClientTimeout(total=30)
    async with aiohttp.ClientSession(timeout=timeout) as session:
        async with session.get(
            "https://api.ipify.org",
            proxy=GATEWAY,
            proxy_auth=AUTH,
        ) as response:
            print(await response.text())


asyncio.run(main())

Scrapy

在 settings.py 里启用中间件：

python
DOWNLOADER_MIDDLEWARES = {
    "scrapy.downloadermiddlewares.httpproxy.HttpProxyMiddleware": 750,
    "myproject.middlewares.NextProxyMiddleware": 751,
}

CONCURRENT_REQUESTS = 32
CONCURRENT_REQUESTS_PER_DOMAIN = 16
DOWNLOAD_TIMEOUT = 30
RETRY_TIMES = 3
RETRY_HTTP_CODES = [429, 500, 502, 503, 504]

中间件按请求分配会话：

python
import base64
import itertools


class NextProxyMiddleware:
    """给每个请求分配一个轮转的 session id，实现受控的 IP 轮换。"""

    GATEWAY = "GATEWAY_HOST:58971"
    USER = "USERNAME"
    PASSWORD = "PASSWORD"
    COUNTRY = "US"
    POOL_SIZE = 16

    def __init__(self) -> None:
        self._slots = itertools.cycle(range(self.POOL_SIZE))

    def process_request(self, request, spider) -> None:
        slot = next(self._slots)
        user = (
            f"{self.USER}-country-{self.COUNTRY}"
            f"-session-slot.{slot}-time-10"
        )
        credentials = base64.b64encode(
            f"{user}:{self.PASSWORD}".encode()
        ).decode()

        request.meta["proxy"] = f"http://{self.GATEWAY}"
        request.headers["Proxy-Authorization"] = f"Basic {credentials}"

从提取接口拿代理池

python
import requests

API = "https://api.example.com/api/v1/proxy/extract"


def fetch_proxy_pool(api_key: str, count: int = 10) -> list[str]:
    response = requests.get(
        API,
        params={
            "apikey": api_key,
            "num": count,
            "type": "json",
            "country": "US",
            "session": "sticky",
            "time": 10,
        },
        timeout=15,
    )
    response.raise_for_status()
    payload = response.json()
    return [f"http://{item['ip']}:{item['port']}" for item in payload["data"]]


# 免密端口，不需要凭据
pool = fetch_proxy_pool("YOUR_API_KEY", count=5)
for proxy in pool:
    print(requests.get(
        "https://api.ipify.org",
        proxies={"http": proxy, "https": proxy},
        timeout=30,
    ).text)

排错清单

现象常见原因
ProxyError: 407凭据写错；proxies 的 value 用了 https://
ProxyError: 400invalid_proxy_parameters：参数格式、顺序或依赖错误
ProxyError: 402流量池或子账号额度不足
ProxyError: 403账号禁用/过期或访问策略拒绝
ProxyError: 429账号并发或建连速率限制
ProxyError: 502出口连接失败
ProxyError: 503没有可用线路或服务暂不可用
ProxyError: 504代理连接建立超时
IP 一直不变连接被复用，或忘了不带 session
IP 该固定却在变time 到期；session 值大小写不一致
SOCKS5 报 Missing dependencies没装 requests[socks]

HTTP / CONNECT 失败可读取 X-NextProxy-Error 与 JSON 的 error.code；目标网站返回的状态和 CONNECT 握手错误需分别处理。完整映射见 错误码对照。

这一类的其他文档

这篇解决你的问题了吗？