如果你在 Python 裡大規模地爬取、爬行或自動化任何東西,目標站台遲早會注意到每個請求都來自同一個 IP,然後把你擋下來。在 Python 中輪換代理能把你的流量分散到許多 IP 位址,讓沒有任何單一 IP 觸發速率限制或招來封鎖。本文展示 requests、httpx 與 aiohttp 的確切程式碼,以及如何在輪換閘道與託管 IP 池之間做選擇。
有兩個橫跨全篇的決策決定了底下的一切。第一:輪換 vs 黏滯工作階段 — 每個請求一個新 IP,或是為多步驟流程固守一個 IP。第二:住宅 vs 機房 IP — 參見 代理類型決策矩陣。把這兩件事弄對,程式碼就是簡單的部分了。
1. 閘道輪換(推薦)。你連到單一端點,供應商在每次新連線時給你一個全新的出口 IP。沒有清單要管理,沒有失效 IP 要清除。JIBAO 的 動態住宅池就是這樣運作:相同的主機與連接埠,每個請求一個新 IP。
2. 自行管理的 IP 清單。你持有一份 host:port 項目的清單,每個請求挑一個。控制權更高,但健康檢查、重試與封鎖追蹤都得自己扛。當你需要已知、穩定的位址時,搭配 專用機房 IP很有用。
最簡單的情況 — 把一個請求透過一個替你輪換的閘道路由出去。使用 socks5h://(結尾的 h 表示 DNS 在代理端解析,可避免洩漏查詢):
import requests
# JIBAO rotating residential gateway: a new exit IP per connection
PROXY = "socks5h://USERNAME:[email protected]:10001"
proxies = {"http": PROXY, "https": PROXY}
for _ in range(5):
r = requests.get("https://httpbin.org/ip", proxies=proxies, timeout=30)
print(r.json()["origin"]) # a different IP each loop
需要讓 requests 講 SOCKS5?安裝額外套件:pip install "requests[socks]"。如果你偏好 HTTP 代理端點,把協定改成 http:// 即可 — 其餘完全相同。
import random
import requests
PROXY_POOL = [
"socks5h://USERNAME-country-us:[email protected]:10001",
"socks5h://USERNAME-country-uk:[email protected]:10001",
"socks5h://USERNAME-country-de:[email protected]:10001",
]
def fetch(url):
proxy = random.choice(PROXY_POOL)
return requests.get(url, proxies={"http": proxy, "https": proxy}, timeout=30)
resp = fetch("https://example.com")
print(resp.status_code)
登入流程、購物車與分頁結果,一旦 IP 在任務中途改變就會壞掉。在使用者名稱加上一個工作階段權杖來釘住一個 IP;重複使用同一個權杖,就能在其生命週期內維持同一個出口 IP:
import requests
SESSION_ID = "task-0001"
PROXY = f"socks5h://USERNAME-session-{SESSION_ID}:[email protected]:10001"
s = requests.Session()
s.proxies = {"http": PROXY, "https": PROXY}
s.post("https://example.com/login", data={"u": "me", "p": "secret"})
s.get("https://example.com/dashboard") # same IP, session intact
httpx 給你一個具備 HTTP/2 與原生非同步的現代用戶端。注意較新版本採用單一的 proxy= 參數:
import httpx
PROXY = "socks5h://USERNAME:[email protected]:10001"
with httpx.Client(proxy=PROXY, timeout=30) as client:
for _ in range(5):
print(client.get("https://httpbin.org/ip").json()["origin"])
httpx 的 SOCKS 支援需要 pip install "httpx[socks]"。要用 HTTP/2,在用戶端加上 http2=True — 它會讓你的流量更像真實瀏覽器,與 TLS 指紋比對搭配得很好。
追求高吞吐量時,就一次發出大量請求。透過閘道的每個連線都會拿到自己的出口 IP,所以並行與輪換是同時發生的:
import asyncio
import aiohttp
from aiohttp_socks import ProxyConnector
PROXY = "socks5://USERNAME:[email protected]:10001"
async def fetch(url):
connector = ProxyConnector.from_url(PROXY)
async with aiohttp.ClientSession(connector=connector) as session:
async with session.get(url, timeout=aiohttp.ClientTimeout(total=30)) as r:
return await r.text()
async def main():
urls = ["https://httpbin.org/ip"] * 20
results = await asyncio.gather(*(fetch(u) for u in urls))
print(f"fetched {len(results)} pages")
asyncio.run(main())
用 pip install aiohttp_socks 安裝 SOCKS 連接器。每個任務都建立一個全新的連接器,讓每個請求都開啟一條新的代理連線。
只有在失敗時用新 IP 重試,輪換才划算。指數退避,並在每次嘗試時輪換工作階段權杖:
import requests
from time import sleep
def fetch_with_retry(url, max_retries=4):
for attempt in range(max_retries):
proxy = f"socks5h://USERNAME-session-r{attempt}:[email protected]:10001"
try:
r = requests.get(url, proxies={"http": proxy, "https": proxy}, timeout=30)
if r.status_code in (403, 429):
raise requests.HTTPError(f"blocked: {r.status_code}")
r.raise_for_status()
return r.text
except requests.RequestException:
sleep(2 ** attempt) # 1s, 2s, 4s, 8s
raise RuntimeError(f"failed after {max_retries} tries: {url}")
大規模執行之前,先確認 IP 真的有改變。打一個回顯端點幾次,檢查 IP 是否不同。要同時確認你的 TLS 指紋沒有出賣你,就把請求送去跑 JIBAO 免費的 JA3/TLS 指紋檢測工具 — 一個輪換的 IP 配上一目了然的 Python 指紋,照樣會被封鎖。
一旦輪換穩固了,你撞上的下一道牆通常是指紋辨識,而不是 IP。如果乾淨的住宅 IP 仍然吃到 403,請閱讀 如何用 curl_cffi 繞過 TLS 指紋辨識以及完整的 Cloudflare 繞過配方。
新用戶註冊即送 500M免費流量,首次儲值另有贈金。優惠限時供應。