Асинхронный веб-скрейпинг с помощью Python и Aiohttp

TL;DR

asyncio улучшает пропускную способность для многих независимых сетевых ожиданий; он не делает парсинг HTML быстрее и не дает разрешения увеличить частоту запросов к сайту.
Переиспользуйте один aiohttp.ClientSession, установите общие и соединительные тайм-ауты, и ограничьте как глобальное, так и пер-хостово конкурентное выполнение.
Семафор ограничивает одновременные задачи, а не запросы в секунду. Добавьте явную синхронизацию по хостам, когда источник этого требует.
Повторяйте только временные ошибки, такие как 429, 502, 503 и 504, с ограниченной задержкой и обработкой Retry-After; не повторяйте слепо каждый ответ 4xx.
Для сайтов с рендерингом JavaScript или рекурсивным обнаружением вызывайте управляемое рендеринг и Crawl API, а не ожидайте, что aiohttp выполнит JavaScript.

Когда следует использовать Asyncio для веб-скрапинга?

Используйте asyncio, когда скрепер проводит большую часть своего времени, ожидая множество разрешенных HTTP-ответов. Один рабочий процесс может управлять несколькими активными запросами без отдельного потока для каждого запроса. Преимущество зависит от задержки источника, разрешенной частоты запросов, стоимости парсинга и хранилища на уровне ниже. Если источник разрешает только один запрос каждые несколько секунд, асинхронный клиент не может законно превратить это в высокочастотный скан. Продукт Nstdata Crawl является управляемым вариантом для шагов получения и преобразования, обсуждаемых здесь.

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

import aiohttp
from bs4 import BeautifulSoup

URLS = ["https://example.com/"]  # Замените на разрешенные URL.
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: {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"Отсутствует заголовок: {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"Достигнут лимит повторных попыток: {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 (contact: 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 для веб-скрейпинга: правильное выполнение параллельных запросов

TL;DR

Когда следует использовать Asyncio для веб-скрапинга?

Популярные статьи

Asyncio для веб-скрейпинга: правильное выполнение параллельных запросов

Веб-скрепинг с Python: Полное руководство 2026 года

Что вам нужно перед началом?

Как конкуренция отличается от ограничения скорости?

Как загружать страницы одновременно с Aiohttp?

Метод 1: Запустите ограниченный извлекатель HTML

Как обрабатывать тысячи URL без создания тысяч задач одновременно?

Метод 2: Используйте ограниченную рабочую очередь

Когда следует использовать Nstdata Crawl вместо Aiohttp?

Метод 3: Делегировать отрисовку и открытие

Почему не удается выполнить параллельные запросы или они возвращают некорректные данные?

Как сделать пример безопасным для производства?

Что следует отслеживать в параллельном скрепере?

Часто задаваемые вопросы