非同步爬取正是代理設定悄悄崩解的地方。用 requests 時，你每次呼叫輪換一個 IP 然後繼續；用 asyncio 時，你突然有 200 個請求同時在飛，每一個都需要自己的 exit、自己的工作階段親和性，以及在目標丟出 403 時自己的重試。搞錯了，你不是把一個 IP 猛敲到被封，就是因為在流程途中輪換而撕碎了每一個登入工作階段。
本指南展示在並行下真正撐得住的模式：httpx 與 aiohttp 的每個請求輪換、一個有上限的工作者 pool、用於有狀態流程的 sticky 工作階段，以及能感知封鎖的重試 — 附上你可以直接貼上的可運作程式碼。
一個同步爬蟲一次只接觸一個代理，所以「每次請求都輪換」是顯然正確的。非同步一次打破三個假設：
httpx.AsyncClient 一個用戶端綁定一個代理，所以要輪換，你就針對每個請求挑選代理，並透過一個以 exit 為鍵的小型用戶端 pool 來路由：
import asyncio, itertools, httpx
PROXIES = [
"socks5h://USERNAME:[email protected]:913",
"socks5h://USERNAME:[email protected]:913",
"socks5h://USERNAME:[email protected]:913",
]
pool = itertools.cycle(PROXIES)
# One reusable client per exit (connection pooling stays intact)
clients = {p: httpx.AsyncClient(proxy=p, timeout=30) for p in PROXIES}
async def fetch(url):
proxy = next(pool)
r = await clients[proxy].get(url)
return r.status_code, r.text
async def main(urls):
results = await asyncio.gather(*(fetch(u) for u in urls))
for client in clients.values():
await client.aclose()
return results每個 exit 重複使用一個用戶端，能讓 HTTP/2 與連線池保持存活，而不是在每次請求都付出一次全新的交握。注意 httpx>=0.28 把用戶端上的 proxies= 改名為 proxy=。
aiohttp 在這裡更簡單 — 單一個 ClientSession 在每個請求接受一個 proxy= 引數，所以你可以行內輪換：
import asyncio, itertools, aiohttp
pool = itertools.cycle(PROXIES) # http:// or socks5h:// (needs aiohttp-socks for SOCKS)
async def fetch(session, url):
proxy = next(pool)
async with session.get(url, proxy=proxy, timeout=aiohttp.ClientTimeout(total=30)) as r:
return r.status, await r.text()
async def main(urls):
async with aiohttp.ClientSession() as session:
return await asyncio.gather(*(fetch(session, u) for u in urls))aiohttp 原生支援 HTTP 代理；對於 socks5h://，加上 aiohttp-socks 並傳入一個 ProxyConnector。
無上限的 gather 是讓你 pool 裡每個 IP 同時被標記的最快方法。為它設上限，讓每個 exit 承載一個人類看起來合理的並行請求數量：
sem = asyncio.Semaphore(10) # at most 10 in flight
async def fetch_capped(session, url):
async with sem:
return await fetch(session, url)經驗法則：把並行量保持在你擁有的不同 exit 數量或以下，這樣你就不會在單一個住宅 IP 上堆疊許多並行請求。
輪換是為了進門；sticky 工作階段是為了留下來。一次登入、一個購物車，任何綁定 cookie 的東西都必須在整段序列中維持一個 exit。把代理釘在每個任務上，而不是每個請求上：
async def run_account(account, proxy):
# Same exit IP for every step of this account's flow
async with httpx.AsyncClient(proxy=proxy, timeout=30) as client:
await client.post("https://target.site/login", data=account.creds)
await client.get("https://target.site/dashboard")
await client.post("https://target.site/cart", json=account.order)
async def main(accounts):
# One sticky exit per account, accounts run concurrently
await asyncio.gather(*(run_account(a, p)
for a, p in zip(accounts, itertools.cycle(PROXIES))))使用一個支援 sticky 住宅工作階段的供應商，好讓同一個 exit 在任務的整個生命週期中被維持，而不是在你不知情下悄悄輪換。
在並行下，你不能信任一個 200 — 反機器人系統會回傳一個帶有挑戰內容的 200。偵測封鎖，並在一個全新的 exit 上重試那個單一失敗的任務：
BLOCK_MARKERS = ("just a moment", "/cdn-cgi/challenge-platform",
"datadome", "px-captcha", "access denied")
def looks_blocked(status, text):
return status in (403, 429, 503) or any(m in text[:4000].lower() for m in BLOCK_MARKERS)
async def fetch_retry(url, attempts=4):
for _ in range(attempts):
proxy = next(pool)
async with httpx.AsyncClient(proxy=proxy, timeout=30) as c:
r = await c.get(url)
if not looks_blocked(r.status_code, r.text):
return r
await asyncio.sleep(0.5)
raise RuntimeError(f"blocked after {attempts} exits: {url}")有一個陷阱非同步並沒有解決：httpx 與 aiohttp 兩者都坐落在 Python 的 TLS 堆疊上，所以它們送出一個沒有任何真實瀏覽器會產生的 JA3。一個完美的輪換 pool 在一個乾淨的住宅 IP 上，仍會在交握時被指紋辨識為自動化。把非同步輪換與 TLS 模仿配對 — 參見 用 curl_cffi 繞過 TLS 指紋偵測，並在 www.jibaoproxy.com/tools/fingerprint.html 驗證你 exit 的 JA3 + ASN。
新用戶註冊即送 500M免費流量，首次儲值另有贈金。優惠限時供應。