Chat with us, powered by LiveChat
EN|中文
Mobile Proxies Logo
mobileproxies.org
企业级代理服务
开始
教程

Python requests SOCKS5 代理实操教程认证、DNS、重试和轮换

完整指南：Python requests 库使用 SOCKS5 代理，包括身份认证、DNS 解析、错误处理和代理轮换实战模式。

2024年10月10日
15分钟阅读

快速上手：最简 SOCKS5 示例

先确认你拿到的确实是 SOCKS5 代理，不是 HTTP 代理，不然下面这段代码会连不上。

import requests

# For credentials with special chars, use URL encoding:
# from urllib.parse import quote
# PROXY = f"socks5h://{quote('user@123')}:{quote('p@ss!')}@host:port"
PROXY = "socks5h://username:password@host:port"  # e.g. socks5h://user:pass@1.2.3.4:1080
proxies = {"http": PROXY, "https": PROXY}

resp = requests.get("https://httpbin.org/ip", proxies=proxies, timeout=20)
resp.raise_for_status()
print(resp.json())

重要提示

  • • 用 pip install "requests[socks]" 安装 SOCKS 支持（会自动装上 PySocks）
  • • socks5h 确保代理来做 DNS 解析。用普通的 socks5 会在本地解析

生产环境推荐这样建 Session：加超时 + 重试（带指数退避）

使用 Session，挂载一个带 urllib3.Retry 的 adapter，高效复用连接：

import os
from urllib.parse import quote
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# Build a socks5h proxy URL (remote DNS)
user = os.getenv("PROXY_USER")
pw = os.getenv("PROXY_PASS", "")
host = os.getenv("PROXY_HOST", "127.0.0.1")
port = int(os.getenv("PROXY_PORT", "1080"))

PROXY = (
    f"socks5h://{quote(user)}:{quote(pw)}@{host}:{port}"
    if user else f"socks5h://{host}:{port}"
)

session = requests.Session()
session.trust_env = False  # optional: ignore env proxies (HTTP_PROXY/HTTPS_PROXY)
session.proxies.update({"http": PROXY, "https": PROXY})

retries = Retry(
    total=5,
    backoff_factor=0.5,  # 0.5, 1, 2, 4... seconds
    status_forcelist=[429, 500, 502, 503, 504],
    allowed_methods=frozenset(["HEAD", "GET", "OPTIONS"]),
    respect_retry_after_header=True,  # explicit for portability with urllib3 v1.26+
)

adapter = HTTPAdapter(max_retries=retries, pool_connections=50, pool_maxsize=50)
session.mount("http://", adapter)
session.mount("https://", adapter)

def fetch(url):
    r = session.get(url, timeout=(5, 30))  # (connect, read)
    r.raise_for_status()
    return r

if __name__ == "__main__":
    print(fetch("https://httpbin.org/ip").json())

实操技巧： Retry 提供指数退避，并且遵守 Retry-After 头（urllib3 v2.0+ 默认开启，v1.26 需要显式设置兼容性）。老项目如果还是 urllib3 1.26.x，要保留 respect_retry_after_header=True 这一行。

环境变量： session.trust_env = False 可以防止 HTTP_PROXY/HTTPS_PROXY 环境变量干扰你显式配置的代理。

⚠️ POST 安全性： 我们从默认重试中去掉了 POST。如果你必须重试 POST 请求，请使用幂等性键并接受重复操作的风险。

socks5 vs socks5h 的区别（以及什么时候会掉坑）

socks5://...

本地 DNS 解析（你的机器先查主机名，然后让代理连接到那个 IP）。

socks5h://...

远程 DNS 解析（代理上解析，防止 DNS 泄露，.onion 和某些服务商必须用这个）。做暗网/.onion、或者供应商要求必须走远程解析的时候，一定用这个写法。

有些代理网络甚至要求远程 DNS（仅主机名目标），所以用 socks5h。

IP 轮换：按需触发 & 轮询模式

移动代理服务商通常提供两种轮换方式：API 触发 IP 变更或内置轮换网关。

方案 1：API 触发轮换（下面以 Coronium.io 做示例，其他供应商换成自己的接口）

很多服务商给你一个重启/轮换端点。你在每批抓取前调用它获取新 IP：

import os
import requests
from urllib.parse import quote
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# Your modem restart token (copy from dashboard.coronium.io)
ROTATION_TOKEN = os.getenv("ROTATION_TOKEN", "19a18981f48fa3907f04c19bdcba8666")
ROTATION_URL = f"https://dashboard.coronium.io/api/v1/modems/restart-by-token/{ROTATION_TOKEN}/"

# Proxy credentials
user = os.getenv("PROXY_USER")
pw = os.getenv("PROXY_PASS", "")
host = os.getenv("PROXY_HOST", "127.0.0.1")
port = int(os.getenv("PROXY_PORT", "1080"))

PROXY = f"socks5h://{quote(user)}:{quote(pw)}@{host}:{port}" if user else f"socks5h://{host}:{port}"

session = requests.Session()
session.trust_env = False
session.proxies.update({"http": PROXY, "https": PROXY})

retries = Retry(
    total=5,
    backoff_factor=0.5,
    status_forcelist=[429, 500, 502, 503, 504],
    allowed_methods=frozenset(["HEAD", "GET", "OPTIONS"]),
    respect_retry_after_header=True,
)
adapter = HTTPAdapter(max_retries=retries, pool_connections=50, pool_maxsize=50)
session.mount("http://", adapter)
session.mount("https://", adapter)

def rotate_ip():
    """Trigger modem restart to get a new IP (实际等待时间看你的供应商，一般30~60秒)."""
    try:
        resp = requests.get(ROTATION_URL, timeout=10)
        resp.raise_for_status()
        print("✓ IP rotation triggered")
        return True
    except Exception as e:
        print(f"⚠ Rotation failed: {e}")
        return False

def fetch(url):
    r = session.get(url, timeout=(5, 30))
    r.raise_for_status()
    return r

if __name__ == "__main__":
    # Rotate before starting
    rotate_ip()
    # Wait for modem to reconnect (adjust timing based on your provider)
    import time
    time.sleep(45)

    # Now scrape with fresh IP
    print(fetch("https://httpbin.org/ip").json())

技巧： 你可以直接从 dashboard.coronium.io 复制你的重启链接 - 里面包含你的唯一 token。需要新 IP 时调用它（例如每 N 个请求或遇到限速后）。

方案 2：多代理轮询

如果你有多个代理，用 itertools.cycle 轮流使用：

import itertools
from urllib.parse import quote

# List of proxies
PROXY_LIST = [
    "socks5h://user1:pass1@host1:1080",
    "socks5h://user2:pass2@host2:1080",
    "socks5h://user3:pass3@host3:1080",
]

proxy_pool = itertools.cycle(PROXY_LIST)

def fetch_with_rotation(url):
    proxy = next(proxy_pool)
    proxies = {"http": proxy, "https": proxy}
    r = requests.get(url, proxies=proxies, timeout=(5, 30))
    r.raise_for_status()
    return r

# Each call uses the next proxy in rotation
for i in range(5):
    print(f"Request {i+1}:", fetch_with_rotation("https://httpbin.org/ip").json())

⚠️ 注意：轮询适合你手上有几条互不影响的代理。要是用那种'同一个IP池里点一下就换IP'的移动代理，就用上面那个按需触发的方案，别来回抢同一条线。

参考资料 & 延伸阅读

需要能稳定跑上面这些代码的 SOCKS5/移动代理？

我们有真实运营商IP的节点，拿来做 Python 自动化刚好。

查看价格开始免费试用