高吞吐量异步语料采集核心代码
采用 Python asyncio + httpx 框架配合每次请求自动轮换 IP 的代理配置,高效实现每分钟上万页面的稳定抓取。
example.py
import asyncio
import httpx
# bytesflows 自动轮换出口:每个全新 TCP/HTTP 请求自动分配全新住宅 IP
PROXY_URL = "http://user-youraccount-rotate:password@proxy.bytesflows.com:31112"
async def fetch_target(client: httpx.AsyncClient, url: str):
try:
response = await client.get(url, timeout=10.0)
print(f"[{response.status_code}] 成功从 {url} 下载 {len(response.text)} 字节")
return response.text
except Exception as e:
print(f"采集报错 {url}: {e}")
async def run_batch_extraction(urls: list[str]):
async with httpx.AsyncClient(proxy=PROXY_URL, verify=False) as client:
tasks = [fetch_target(client, u) for u in urls]
return await asyncio.gather(*tasks)
if __name__ == "__main__":
target_list = ["https://httpbin.org/html"] * 20
asyncio.run(run_batch_extraction(target_list))