aiohttp 2026: 构建每秒处理10K+请求的高性能异步网页抓取器 — Python指南

pip install aiohttp==3.11.0

# 包含加速组件（生产环境推荐）
pip install aiohttp[speedups]==3.11.0

# 安装抓取所需的附加工具
pip install aiohttp==3.11.0 aiofiles==24.1.0 beautifulsoup4==4.12.3 lxml==5.3.0

import aiohttp
import asyncio
import sys

print(f"aiohttp version: {aiohttp.__version__}")
print(f"Python version: {sys.version}")

async def check():
    async with aiohttp.ClientSession() as session:
        async with session.get("https://httpbin.org/get") as resp:
            data = await resp.json()
            print(f"Status: {resp.status}")
            print(f"Response keys: {list(data.keys())}")

asyncio.run(check())

import aiohttp
import asyncio

urls = [
    "https://httpbin.org/get?param=1",
    "https://httpbin.org/get?param=2",
    "https://httpbin.org/get?param=3",
]

async def fetch(session, url):
    async with session.get(url) as response:
        return await response.json()

async def main():
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        for r in results:
            print(r["args"])

asyncio.run(main())

import aiohttp
import asyncio
from bs4 import BeautifulSoup

async def scrape_titles(session, urls):
    """从多个 URL 并发提取页面标题。"""
    titles = []
    for url in urls:
        try:
            async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
                html = await resp.text()
                soup = BeautifulSoup(html, "lxml")
                title = soup.find("title")
                titles.append({"url": url, "title": title.text if title else "N/A"})
        except Exception as e:
            titles.append({"url": url, "title": f"Error: {e}"})
    return titles

async def main():
    urls = ["https://example.com", "https://httpbin.org/html"]
    async with aiohttp.ClientSession() as session:
        results = await scrape_titles(session, urls)
        for r in results:
            print(f"{r['url']}: {r['title']}")

asyncio.run(main())

import aiohttp
import asyncio
from lxml import html as lh

async def extract_links(session, url):
    """使用 lxml 从页面提取所有 href 链接。"""
    async with session.get(url) as resp:
        text = await resp.text()
        tree = lh.fromstring(text)
        links = tree.xpath("//a/@href")
        return [l for l in links if l.startswith("http")]

async def main():
    async with aiohttp.ClientSession() as session:
        links = await extract_links(session, "https://example.com")
        print(f"Found {len(links)} external links")

asyncio.run(main())

import aiohttp
import aiofiles
import asyncio
import json

async def scrape_and_save(session, url, filepath):
    """抓取数据并异步写入磁盘。"""
    async with session.get(url) as resp:
        data = await resp.json()
        async with aiofiles.open(filepath, "w") as f:
            await f.write(json.dumps(data, indent=2))

async def main():
    async with aiohttp.ClientSession() as session:
        await scrape_and_save(
            session,
            "https://httpbin.org/json",
            "/tmp/scraped_data.json"
        )

asyncio.run(main())

import aiohttp
import aiosqlite
import asyncio

async def scrape_to_db(session, db, url):
    """异步将抓取数据存储到 SQLite。"""
    async with session.get(url) as resp:
        data = await resp.json()
        await db.execute(
            "INSERT INTO scraped (url, data) VALUES (?, ?)",
            (url, json.dumps(data))
        )
        await db.commit()

async def main():
    async with aiosqlite.connect("scraped.db") as db:
        await db.execute("CREATE TABLE IF NOT EXISTS scraped (url TEXT, data TEXT)")
        async with aiohttp.ClientSession() as session:
            await scrape_to_db(session, db, "https://httpbin.org/json")

asyncio.run(main())

import aiohttp
import asyncio

PROXY_URL = "http://username:password@proxy.webshare.io:80"

async def fetch_with_proxy(session, url):
    """通过 WebShare 轮换代理路由请求。"""
    async with session.get(url, proxy=PROXY_URL) as resp:
        return await resp.text()

async def main():
    connector = aiohttp.TCPConnector(limit=100, limit_per_host=10)
    async with aiohttp.ClientSession(connector=connector) as session:
        html = await fetch_with_proxy(session, "https://httpbin.org/ip")
        print(html[:200])

asyncio.run(main())

import aiohttp

connector = aiohttp.TCPConnector(
    limit=200,              # 总并发连接数
    limit_per_host=20,      # 每主机连接数（尊重服务器！）
    ttl_dns_cache=300,      # DNS 缓存 TTL（秒）
    use_dns_cache=True,     # 启用 DNS 缓存
    enable_cleanup_closed=True,
    force_close=False,      # 保持连接存活
)

timeout = aiohttp.ClientTimeout(
    total=30,               # 每个请求的总超时
    connect=5,              # TCP 连接超时
    sock_read=15,           # 套接字读取超时
)

session = aiohttp.ClientSession(
    connector=connector,
    timeout=timeout,
    headers={"User-Agent": "MyBot/1.0"},
)

import aiohttp
import asyncio

async def bounded_fetch(session, url, semaphore):
    """使用信号量限制并发请求数。"""
    async with semaphore:
        async with session.get(url) as resp:
            return await resp.text()

async def main():
    semaphore = asyncio.Semaphore(50)  # 最大 50 个并发请求
    urls = [f"https://httpbin.org/get?i={i}" for i in range(500)]
    
    connector = aiohttp.TCPConnector(limit=100)
    async with aiohttp.ClientSession(connector=connector) as session:
        tasks = [bounded_fetch(session, url, semaphore) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        successes = sum(1 for r in results if not isinstance(r, Exception))
        print(f"Successful: {successes}/500")

asyncio.run(main())

import aiohttp
import asyncio
import random

async def fetch_with_retry(session, url, max_retries=3):
    """指数退避重试失败的请求。"""
    for attempt in range(max_retries):
        try:
            async with session.get(url) as resp:
                if resp.status == 200:
                    return await resp.json()
                elif resp.status in (429, 503, 502):
                    wait = (2 ** attempt) + random.uniform(0, 1)
                    await asyncio.sleep(wait)
                else:
                    resp.raise_for_status()
        except (aiohttp.ClientError, asyncio.TimeoutError) as e:
            if attempt == max_retries - 1:
                raise
            await asyncio.sleep(2 ** attempt)
    return None

async def main():
    async with aiohttp.ClientSession() as session:
        data = await fetch_with_retry(session, "https://httpbin.org/json")
        print(data)

asyncio.run(main())

import aiohttp
import asyncio

async def websocket_scraper():
    """从 WebSocket 端点抓取实时数据。"""
    async with aiohttp.ClientSession() as session:
        async with session.ws_connect("wss://echo.websocket.org") as ws:
            await ws.send_str("Hello Server")
            
            async for msg in ws:
                if msg.type == aiohttp.WSMsgType.TEXT:
                    print(f"Received: {msg.data}")
                    if "done" in msg.data.lower():
                        await ws.close()
                        break
                elif msg.type == aiohttp.WSMsgType.ERROR:
                    print(f"WebSocket error: {ws.exception()}")
                    break

asyncio.run(websocket_scraper())

# Dockerfile
FROM python:3.12-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY scraper.py .
CMD ["python", "scraper.py"]

# docker-compose.yml
version: "3.8"
services:
  scraper:
    build: .
    restart: unless-stopped
    environment:
      - PYTHONUNBUFFERED=1
    deploy:
      resources:
        limits:
          memory: 2G
    logging:
      driver: "json-file"
      options:
        max-size: "100m"
        max-file: "3"

import aiohttp
import asyncio
from prometheus_client import Counter, Histogram, start_http_server

REQUEST_COUNT = Counter("scraper_requests_total", "总请求数", ["status"])
REQUEST_DURATION = Histogram("scraper_request_duration_seconds", "请求耗时")

async def monitored_fetch(session, url):
    with REQUEST_DURATION.time():
        try:
            async with session.get(url) as resp:
                REQUEST_COUNT.labels(status=str(resp.status)).inc()
                return await resp.text()
        except Exception as e:
            REQUEST_COUNT.labels(status="error").inc()
            raise

# 在端口 9090 启动指标服务器
start_http_server(9090)