Python 异步爬虫流水线：限制并发、控制内存，并把结果持续写入文件

export DOC_URL_1='<paste-the-actual-doc-url-1>'
export DOC_URL_2='<paste-the-actual-doc-url-2>'
export DOC_URL_3='<paste-the-actual-doc-url-3>'

URL 清单 → 有界任务队列 → 4 个下载协程
                       ↓
                  有界页面队列 → 1 个解析协程
                                      ↓
                                 有界结果队列 → 1 个文件写入协程

python -m venv .venv
source .venv/bin/activate
python -m pip install "httpx==0.28.1" "beautifulsoup4==4.13.4"

import asyncio
import os
import json
from pathlib import Path
 
import httpx
from bs4 import BeautifulSoup
 
URLS = [
    os.environ["DOC_URL_1"],
    os.environ["DOC_URL_2"],
    os.environ["DOC_URL_3"],
]
WORKERS = 4
MAX_BYTES = 2 * 1024 * 1024
OUTPUT = Path("pages.jsonl")
 
 
class PageTooLarge(Exception):
    pass
 
 
async def produce(jobs):
    for url in URLS:
        await jobs.put(url)
    for _ in range(WORKERS):
        await jobs.put(None)
 
 
async def fetch(client, jobs, pages):
    while True:
        url = await jobs.get()
        if url is None:
            return
        try:
            async with client.stream("GET", url) as response:
                response.raise_for_status()
                media = response.headers.get("content-type", "").lower()
                if "text/html" not in media:
                    raise ValueError("unexpected_content_type")
                body = bytearray()
                async for chunk in response.aiter_bytes():
                    if len(body) + len(chunk) > MAX_BYTES:
                        raise PageTooLarge("response_exceeds_2_mib")
                    body.extend(chunk)
                record = {
                    "url": str(response.url),
                    "requested_url": url,
                    "status": "downloaded",
                    "body": bytes(body),
                    "encoding": response.encoding,
                }
        except (httpx.HTTPError, PageTooLarge, ValueError) as exc:
            record = {
                "url": url,
                "requested_url": url,
                "status": "failed",
                "stage": "download",
                "error_type": type(exc).__name__,
                "error": str(exc),
            }
        await pages.put(record)
 
 
async def parse(pages, results):
    while True:
        record = await pages.get()
        if record is None:
            await results.put(None)
            return
        if record["status"] == "failed":
            await results.put(record)
            continue
        soup = BeautifulSoup(
            record.pop("body"),
            "html.parser",
            from_encoding=record.pop("encoding"),
        )
        title = soup.title.get_text(" ", strip=True) if soup.title else ""
        if not title:
            record.update(
                status="failed", stage="parse",
                error_type="MissingTitle", error="title_is_empty",
            )
        else:
            record.update(status="ok", title=title)
        await results.put(record)
 
 
async def write(results):
    written = 0
    with OUTPUT.open("x", encoding="utf-8", newline="\n") as handle:
        try:
            while True:
                record = await results.get()
                if record is None:
                    break
                handle.write(json.dumps(record, ensure_ascii=False) + "\n")
                written += 1
                if written % 10 == 0:
                    handle.flush()
        finally:
            handle.flush()
            print("written_records=" + str(written))
 
 
async def finish_downloads(tasks, pages):
    await asyncio.gather(*tasks)
    await pages.put(None)
 
 
async def main():
    jobs = asyncio.Queue(maxsize=16)
    pages = asyncio.Queue(maxsize=8)
    results = asyncio.Queue(maxsize=16)
    limits = httpx.Limits(
        max_connections=WORKERS, max_keepalive_connections=WORKERS
    )
    async with httpx.AsyncClient(
        timeout=httpx.Timeout(20.0),
        limits=limits,
        follow_redirects=True,
        headers={"User-Agent": "DocumentPipelineTutorial/1.0"},
    ) as client:
        async with asyncio.TaskGroup() as group:
            group.create_task(produce(jobs))
            downloads = [
                group.create_task(fetch(client, jobs, pages))
                for _ in range(WORKERS)
            ]
            group.create_task(parse(pages, results))
            group.create_task(write(results))
            group.create_task(finish_downloads(downloads, pages))
 
 
if __name__ == "__main__":
    asyncio.run(main())

python pipeline.py

{"url":"https://example.16yun.cn/3/tutorial/introduction.html","requested_url":"https://example.16yun.cn/3/tutorial/introduction.html","status":"ok","title":"An Informal Introduction to Python — Python documentation"}

import json
from collections import Counter
 
with open("pages.jsonl", encoding="utf-8") as handle:
    rows = [json.loads(line) for line in handle]
print("records:", len(rows))
print("statuses:", dict(Counter(row["status"] for row in rows)))
print("inputs:", len({row["requested_url"] for row in rows}))
for row in rows:
    if row["status"] != "ok":
        print(row["requested_url"], row["stage"], row["error_type"])