5.5 示例项目：优化网络爬虫

# 依赖：
# pip install requests beautifulsoup4

import time
from typing import List, Dict
import requests
from bs4 import BeautifulSoup

HEADERS = {"User-Agent": "MasteringClaudeCode/1.0 (+https://example.org)"}

def fetch(url: str, timeout: float = 10.0) -> str:
    """同步获取一个 URL 并返回文本内容。"""
    resp = requests.get(url, headers=HEADERS, timeout=timeout)
    resp.raise_for_status()
    return resp.text

def parse(html: str) -> Dict[str, str]:
    """提取简单字段：<title> 和第一个 <h1>（如果存在）。"""
    soup = BeautifulSoup(html, "html.parser")
    title = (soup.title.string or "").strip() if soup.title else ""
    h1 = (soup.find("h1").get_text(strip=True)) if soup.find("h1") else ""
    return {"title": title, "h1": h1}

def scrape(urls: List[str]) -> List[Dict[str, str]]:
    results = []
    for url in urls:
        html = fetch(url)
        data = parse(html)
        data["url"] = url
        results.append(data)
        time.sleep(0.2) # 礼貌性延迟
    return results

if __name__ == "__main__":
    URLS = [
        "https://example.com",
        "https://httpbin.org/html",
        "https://www.iana.org/domains/reserved",
    ]
    start = time.time()
    items = scrape(URLS)
    elapsed = time.time() - start
    for item in items:
        print(f"{item['url']}\n title: {item['title']}\n h1: {item['h1']}\n")
    print(f"Baseline elapsed: {elapsed:.3f}s")

这个基础版本简单且正确，但它在每次网络请求时都会阻塞。随着 URL 列表的增长，总耗时与延迟呈线性关系，并且任何瞬时错误都会中断处理。

优化版：并发、连接池化且健壮的爬虫

# optimized_scraper.py
# 依赖：
# pip install aiohttp aiodns beautifulsoup4 cchardet orjson
# （aiodns 为可选；在某些平台上可加速 DNS 解析）
import asyncio
from typing import List, Dict, Optional, Tuple
from contextlib import asynccontextmanager
import random
import time
import aiohttp
from bs4 import BeautifulSoup

HEADERS = {"User-Agent": "MasteringClaudeCode/1.0 (+https://example.org)"}

# 有界并发可避免压垮远端服务器和本地机器。
MAX_CONCURRENCY = 10
REQUEST_TIMEOUT = aiohttp.ClientTimeout(total=12, connect=5, sock_connect=5, sock_read=10)

@asynccontextmanager
async def make_session() -> aiohttp.ClientSession:
    connector = aiohttp.TCPConnector(limit=MAX_CONCURRENCY, ttl_dns_cache=300)
    async with aiohttp.ClientSession(headers=HEADERS, timeout=REQUEST_TIMEOUT, connector=connector) as session:
        yield session

async def fetch(session: aiohttp.ClientSession, url: str, attempt: int = 1, max_attempts: int = 3) -> str:
    """
    获取一个 URL，在瞬时错误时采用指数退避。
    若所有尝试都失败，则抛出最后一个异常。
    """
    try:
        async with session.get(url) as resp:
            # 对非 2xx 响应快速失败
            if resp.status >= 400:
                raise aiohttp.ClientResponseError(
                    request_info=resp.request_info,
                    history=resp.history,
                    status=resp.status,
                    message=f"HTTP {resp.status}",
                    headers=resp.headers,
                )
            # 高效解码；必要时依赖 aiohttp 的 chardet
            return await resp.text()
    except (aiohttp.ClientError, asyncio.TimeoutError) as e:
        if attempt >= max_attempts:
            raise
        # 带抖动的指数退避
        backoff = (2 ** (attempt - 1)) + random.uniform(0, 0.25)
        await asyncio.sleep(backoff)
        return await fetch(session, url, attempt + 1, max_attempts)

def parse_fast(html: str) -> Dict[str, str]:
    """
    优先使用 BeautifulSoup 的 'lxml' 解析器以提升速度，
    不可用时回退到 'html.parser'。
    """
    try:
        soup = BeautifulSoup(html, "lxml")
    except Exception:
        soup = BeautifulSoup(html, "html.parser")
    title = (soup.title.string or "").strip() if soup.title else ""
    h1_tag = soup.find("h1")
    h1 = h1_tag.get_text(strip=True) if h1_tag else ""
    return {"title": title, "h1": h1}

async def scrape_one(semaphore: asyncio.Semaphore, session: aiohttp.ClientSession, url: str) -> Tuple[str, Optional[Dict[str, str]], Optional[str]]:
    async with semaphore:
        try:
            html = await fetch(session, url)
            data = parse_fast(html)
            data["url"] = url
            return url, data, None
        except Exception as e:
            return url, None, f"{type(e).__name__}: {e}"

async def scrape(urls: List[str]) -> List[Dict[str, str]]:
    semaphore = asyncio.Semaphore(MAX_CONCURRENCY)
    async with make_session() as session:
        tasks = [scrape_one(semaphore, session, url) for url in urls]
        results = []
        for coro in asyncio.as_completed(tasks):
            url, data, err = await coro
            if err:
                results.append({"url": url, "error": err, "title": "", "h1": ""})
            else:
                results.append(data)
        return results

if __name__ == "__main__":
    URLS = [
        "https://example.com",
        "https://httpbin.org/html",
        "https://www.iana.org/domains/reserved",
        # 在此处添加更多 URL 以观察并发效果
    ]
    started = time.time()
    items = asyncio.run(scrape(URLS))
    elapsed = time.time() - started
    for item in items:
        if item.get("error"):
            print(f"{item['url']}\n error: {item['error']}\n")
        else:
            print(f"{item['url']}\n title: {item['title']}\n h1: {item['h1']}\n")
    print(f"Optimized elapsed: {elapsed:.3f}s with concurrency={MAX_CONCURRENCY}")

优化版的关键改进：

• 通过信号量和连接器限制实现有界并发，既保持礼貌访问，又保护资源。
• 通过 TCPConnector 实现连接池化，减少大量请求间的握手开销。
• 超时与退避机制可避免请求卡死，并能从容应对瞬时失败，而不会让整个流程崩溃。
• 更快的解析会自动优先选用 lxml，同时保持对标准解析器的兼容性。
• 结构化的结果统一捕获成功与错误，使下游处理更简单。

你从一个正确的同步爬虫出发，通过深入思考时间究竟消耗在哪里，逐步将其演进为一个并发、连接池化且具备容错能力的管道，同时保持了良好的可读性和可修改性。Claude Code 在此工作流中的价值在于：能够解释每次变更为何重要、预测其影响，并在性能提升的同时保持设计的整洁性。在下一节中，你将把这些技术推广为数据采集服务可复用的优化清单：测量瓶颈、选择合适的并发模型，并验证提速不以牺牲正确性或可维护性为代价。