Lấy dữ liệu web bất đồng bộ với Python và Aiohttp

TL;DR

asyncio cải thiện thông lượng cho nhiều chờ đợi mạng độc lập; nó không làm cho việc phân tích HTML nhanh hơn hoặc cấp phép để tăng tỷ lệ yêu cầu của một trang web.
Tái sử dụng một aiohttp.ClientSession, đặt total và connection timeouts, và ràng buộc cả concurrency toàn cục và theo từng host.
Một semaphore giới hạn các tác vụ đồng thời, không phải là yêu cầu mỗi giây. Thêm tốc độ rõ ràng theo từng host khi nguồn yêu cầu điều đó.
Thử lại chỉ những thất bại tạm thời như 429, 502, 503, và 504, với một độ trễ giới hạn và xử lý Retry-After; không thử lại mù quáng mọi phản hồi 4xx.
Đối với các trang web render bằng JavaScript hoặc khám phá đệ quy, gọi một API rendering và Crawl đã được quản lý thay vì mong đợi aiohttp thực thi JavaScript.

Khi Nào Nên Sử Dụng Asyncio ChO Web Scraping?

Sử dụng asyncio khi scraper dành phần lớn thời gian của nó để chờ đợi nhiều phản hồi HTTP được phép. Một worker có thể quản lý nhiều yêu cầu đang thực hiện mà không cần một thread cho mỗi yêu cầu. Lợi ích phụ thuộc vào độ trễ nguồn, tỷ lệ yêu cầu được phép, chi phí phân tích, và lưu trữ downstream. Nếu nguồn chỉ cho phép một yêu cầu mỗi vài giây, một client không đồng bộ không thể hợp pháp biến nó thành một crawl với tỷ lệ cao. Sản phẩm Nstdata Crawl là một lựa chọn được quản lý cho các bước thu thập và chuyển đổi được thảo luận ở đây.

Tài liệu giải thích cách các coroutine được lên lịch. Hướng dẫn khuyến nghị tái sử dụng một session thay vì tạo một cái cho mỗi yêu cầu. Một session giữ một pool kết nối và cấu hình chia sẻ. Sử dụng hướng dẫn này chỉ cho các nguồn bạn được phép thu thập.

import asyncio
import time
from collections import defaultdict
from urllib.parse import urlparse

import aiohttp
from bs4 import BeautifulSoup

URLS = ["https://example.com/"]  # Thay thế bằng các URL được phê duyệt.
MAX_CONCURRENT = 8
MAX_PER_HOST = 3
MIN_HOST_INTERVAL = 1.0
RETRYABLE = {429, 502, 503, 504}


class HostPacer:
    def __init__(self, interval: float):
        self.interval = interval
        self.locks = defaultdict(asyncio.Lock)
        self.next_start = defaultdict(float)

    async def wait(self, host: str):
        async with self.locks[host]:
            delay = self.next_start[host] - time.monotonic()
            if delay > 0:
                await asyncio.sleep(delay)
            self.next_start[host] = time.monotonic() + self.interval


def retry_delay(header, attempt: int) -> float:
    if header and header.isdigit():
        return min(float(header), 60.0)
    return min(2 ** attempt, 30.0)


async def fetch(session, pacer, semaphore, url):
    host = urlparse(url).hostname
if not host or urlparse(url).scheme not in {"http", "https"}:
        raise ValueError(f"URL không được hỗ trợ: {url}")

    for attempt in range(3):
        await pacer.wait(host)
        try:
            async with semaphore:
                async with session.get(url) as response:
                    if response.status in RETRYABLE:
                        delay = retry_delay(response.headers.get("Retry-After"), attempt)
                    else:
                        response.raise_for_status()
                        html = await response.text()
                        soup = BeautifulSoup(html, "html.parser")
                        title = soup.title.get_text(" ", strip=True) if soup.title else None
                        if not title:
                            raise ValueError(f"Thiếu tiêu đề: {url}")
                        return {"url": str(response.url), "status": response.status, "title": title}
        except (aiohttp.ClientError, asyncio.TimeoutError):
            if attempt == 2:
                raise
            delay = min(2 ** attempt, 30.0)

        if attempt == 2:
            raise RuntimeError(f"Đạt giới hạn thử lại: {url}")
        await asyncio.sleep(delay)


async def main():
    timeout = aiohttp.ClientTimeout(total=30, connect=10)
    connector = aiohttp.TCPConnector(limit=MAX_CONCURRENT, limit_per_host=MAX_PER_HOST)
    headers = {"User-Agent": "ExampleResearchBot/1.0 (liên hệ: team@example.org)"}
    pacer = HostPacer(MIN_HOST_INTERVAL)
    semaphore = asyncio.Semaphore(MAX_CONCURRENT)
    async with aiohttp.ClientSession(timeout=timeout, connector=connector, headers=headers) as session:
        results = await asyncio.gather(
            *(fetch(session, pacer, semaphore, url) for url in URLS),
            return_exceptions=True,
        )
    for url, result in zip(URLS, results):
        print(url, result)


if __name__ == "__main__":
    asyncio.run(main())

Asyncio cho Web Scraping: Các Yêu Cầu Đồng Thời Được Thực Hiện Đúng Cách

TL;DR

Khi Nào Nên Sử Dụng Asyncio ChO Web Scraping?

Bai viet pho bien

Asyncio cho Web Scraping: Các Yêu Cầu Đồng Thời Được Thực Hiện Đúng Cách

Web Scraping with Python: Hướng Dẫn Toàn Diện 2026

Bạn Cần Gì Trước Khi Bắt Đầu?

Độ Đồng Thời Khác Nhau Như Thế Nào So Với Giới Hạn Tỷ Lệ?

Làm Thế Nào Để Bạn Fetch Các Trang ĐỒNG THỜI Với Aiohttp?

Phương Pháp 1: Chạy Một HTML Fetcher Có Giới Hạn

Bạn Xử Lý Hàng Nghìn URL Như Thế Nào Mà Không Tạo Ra Hàng Nghìn Nhiệm Vụ Cùng Một Lúc?

Phương pháp 2: Sử Dụng Hàng Đợi Công Việc Giới Hạn

Khi Nào Bạn Nên Sử Dụng Nstdata Crawl Thay Vì Aiohttp?

Phương pháp 3: Giao Nhiệm Vụ Rendering và Phát Hiện

Tại Sao Các Yêu Cầu Đối Chứng Trùng Lặp Lại Thất Bại Hoặc Trả Về Dữ Liệu Xấu?

Làm Thế Nào Để Bạn Giúp Ví Dụ An Toàn Đối Với Sản Xuất?

Bạn Nên Giám Sát Những Gì Trong Một Trình Quét Đối Chứng?

Câu Hỏi Thường Gặp