비동기 스크래핑은 프록시 설정이 조용히 무너지는 지점입니다. requests로는 호출마다 IP 하나를 회전시키고 넘어가면 그만이지만, asyncio를 쓰면 갑자기 200개의 요청이 동시에 떠 있고, 각각 자기만의 출구, 자기만의 세션 어피니티, 그리고 대상이 403을 던질 때를 위한 자기만의 재시도가 필요해집니다. 이를 잘못 다루면 IP 하나를 두들겨 차단당하거나, 흐름 중간에 회전을 일으켜 모든 로그인 세션을 망가뜨리게 됩니다.
이 가이드는 동시성 환경에서 실제로 버텨내는 패턴들을 보여줍니다. httpx와 aiohttp의 요청별 회전, 제한된 워커 풀, 상태 유지 흐름을 위한 sticky 세션, 그리고 차단을 인식하는 재시도까지 — 그대로 붙여넣을 수 있는 동작하는 코드와 함께 다룹니다.
동기식 스크래퍼는 한 번에 프록시 하나만 건드리므로 "요청마다 회전"은 자명하게 옳습니다. 비동기는 세 가지 전제를 한꺼번에 깨뜨립니다:
httpx.AsyncClient는 클라이언트당 프록시 하나를 바인딩하므로, 회전하려면 요청마다 프록시를 선택하고 출구별로 키를 지정한 작은 클라이언트 풀을 통해 라우팅합니다:
import asyncio, itertools, httpx
PROXIES = [
"socks5h://USERNAME:PASSWORD@us.jibaoproxy.com:913",
"socks5h://USERNAME:PASSWORD@de.jibaoproxy.com:913",
"socks5h://USERNAME:PASSWORD@jp.jibaoproxy.com: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출구별로 클라이언트 하나를 재사용하면 매 요청마다 새 핸드셰이크 비용을 치르는 대신 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는 풀의 모든 IP를 한꺼번에 플래그 처리당하게 만드는 가장 빠른 방법입니다. 각 출구가 사람으로 보일 만한 수의 병렬 요청만 감당하도록 제한하세요:
sem = asyncio.Semaphore(10) # at most 10 in flight
async def fetch_capped(session, url):
async with sem:
return await fetch(session, url)경험칙: 동시성을 보유한 고유 출구 수와 같거나 그 이하로 유지해서, 단일 주거용 IP에 많은 병렬 요청을 쌓지 않도록 하세요.
회전은 들어가기 위한 것이고, sticky 세션은 머물러 있기 위한 것입니다. 로그인, 장바구니, 쿠키에 묶인 모든 것은 시퀀스 전체 동안 하나의 출구를 유지해야 합니다. 요청별이 아니라 작업별로 프록시를 고정하세요:
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 주거용 세션을 지원하는 공급자를 사용해서, 같은 출구가 작업 수명 내내 유지되고 모르는 사이에 회전되지 않도록 하세요.
동시성 환경에서는 200을 믿을 수 없습니다 — 안티봇 시스템은 챌린지 본문과 함께 200을 반환합니다. 차단을 감지하고 실패한 그 하나의 작업만 새 출구에서 재시도하세요:
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를 보냅니다. 깨끗한 주거용 IP 위에 완벽한 회전 풀이 있어도 핸드셰이크 단계에서 자동화로 지문 인식됩니다. 비동기 회전을 TLS 임퍼소네이션과 함께 사용하세요 — curl_cffi로 TLS 지문 우회하기를 참고하고, 출구의 JA3 + ASN을 www.jibaoproxy.com/tools/fingerprint.html에서 확인하세요.
신규 사용자는 가입 시 500MB를 받고, 첫 충전 시 추가 보너스를 받습니다. 기간 한정 혜택입니다.