非同期スクレイピングは、プロキシのセットアップがひそかに崩壊する場所です。requests なら 1 回の呼び出しにつき IP を 1 つローテーションして次に進めばよいのですが、asyncio では突然 200 個のリクエストが同時に飛び、それぞれが自分の exit、自分のセッションアフィニティ、そしてターゲットが 403 を投げたときの自分のリトライを必要とします。これを誤れば、1 つの IP を BAN まで叩き続けるか、フロー途中でローテーションしてすべてのログインセッションを破り捨てるかのどちらかになります。
本ガイドは、並行度のもとで実際に持ちこたえるパターンを示します：httpx と aiohttp でのリクエスト単位ローテーション、上限付きのワーカープール、ステートフルなフローのための sticky セッション、そしてブロックを意識したリトライ——貼り付けてすぐ使える動作するコード付きで。
同期スクレイパーは一度に 1 つのプロキシしか触らないので、「リクエストごとにローテーション」は自明に正しいです。非同期は 3 つの前提を一度に壊します：
httpx.AsyncClient はクライアントごとに 1 つのプロキシをバインドするので、ローテーションするにはリクエストごとにプロキシを選び、exit でキー付けした小さなクライアントプールを経由してルーティングします：
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 resultsexit ごとに 1 つのクライアントを再利用することで、毎リクエストで新しいハンドシェイクを払う代わりに、HTTP/2 とコネクションプーリングを生かしたままにできます。なお httpx>=0.28 はクライアントの proxies= を proxy= にリネームしました。
aiohttp はここではよりシンプルです——1 つの 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 は、プール内のすべての 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 の数以下に保ってください。そうすれば 1 つの residential IP に多数の並行リクエストを積み上げずに済みます。
ローテーションは入るためのもの、sticky セッションは留まるためのものです。ログイン、カート、cookie に縛られるものは何であれ、シーケンス全体で 1 つの 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 な residential セッションをサポートするプロバイダーを使い、同じ exit がタスクの寿命のあいだ保持され、あなたの知らぬ間にローテーションされないようにしてください。
並行度のもとでは 200 を信用できません——anti-bot システムはチャレンジのボディを付けて 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}")非同期が解決しない 1 つの落とし穴：httpx も aiohttp もどちらも Python の TLS スタックの上に乗っているので、本物のブラウザが決して生成しない JA3 を送ります。クリーンな residential IP 上の完璧なローテーションプールでも、ハンドシェイクで自動化としてフィンガープリントされてしまいます。非同期ローテーションを TLS なりすましと組み合わせてください——curl_cffi で TLS フィンガープリントを回避を参照し、www.jibaoproxy.com/tools/fingerprint.html で exit の JA3 + ASN を検証してください。
新規ユーザーは登録時に500MBを獲得、さらに初回チャージにボーナスが付きます。期間限定オファーです。