Асинхронный парсинг на Python: aiohttp и asyncio

Когда нужно скачать тысячи или десятки тысяч страниц, синхронный код упирается в стену: каждый запрос ждёт ответа, прежде чем начнётся следующий. Потоки помогают, но тратят память и упираются в накладные расходы. Асинхронность решает задачу элегантнее — один поток держит тысячи одновременных соединений, переключаясь между ними, пока они ждут сеть.

Эта статья — продвинутое продолжение обзорного материала «Парсинг на Python». Базовые приёмы (библиотеки, кодировки) разобраны там; здесь — как масштабировать парсинг через asyncio.

Оглавление

1. Почему async быстрее для парсинга
2. Как забираем страницу: aiohttp
3. Парсинг содержимого в async-коде
4. Кириллица в асинхронном парсинге
5. Управление параллельностью: семафоры
6. Прокси
7. Парсинг через TOR
8. HTTPS/SSL
9. Работа с cookie
10. Статус ответа и заголовки
11. Очереди: asyncio.Queue
12. httpx как альтернатива
13. Плюсы и минусы

1. Почему async быстрее для парсинга

Парсинг — задача I/O-bound: 99% времени программа просто ждёт ответа сервера. В синхронном коде это ожидание простаивает. Асинхронность позволяет, пока один запрос ждёт, запускать сотни других.

• Синхронно: 1000 страниц по 0.5 c = ~500 секунд.
• Асинхронно (100 одновременно): те же 1000 страниц = ~5 секунд.

В отличие от потоков, корутины почти ничего не стоят по памяти — десятки тысяч одновременных задач в одном потоке реальны. Сравните с многопоточным подходом из хаба: async масштабируется заметно выше.

2. Как забираем страницу: aiohttp

aiohttp — стандартный асинхронный HTTP-клиент. Ключевой принцип: один ClientSession на всю программу (переиспользует соединения), множество одновременных запросов через asyncio.gather.

import asyncio
import aiohttp

async def fetch(session, url):
    async with session.get(url, timeout=aiohttp.ClientTimeout(total=15)) as resp:
        return await resp.text()

async def main(urls):
    async with aiohttp.ClientSession(headers={"User-Agent": "MyBot/1.0"}) as session:
        tasks = [fetch(session, url) for url in urls]
        pages = await asyncio.gather(*tasks, return_exceptions=True)
        return pages

urls = [f"https://example.com/page/{i}" for i in range(1, 1001)]
results = asyncio.run(main(urls))

return_exceptions=True важно: одна упавшая задача не уронит весь gather, а вернётся как объект исключения, который вы обработаете.

3. Парсинг содержимого в async-коде

Важный нюанс: сам разбор HTML (BeautifulSoup, lxml) — синхронная CPU-операция. Если HTML тяжёлый, парсинг блокирует event loop и сводит на нет выигрыш от async. Лёгкие страницы можно парсить прямо в корутине:

from bs4 import BeautifulSoup

async def fetch_and_parse(session, url):
    async with session.get(url) as resp:
        html = await resp.text()
    soup = BeautifulSoup(html, "lxml")     # для лёгких страниц — ок
    return soup.find("h1").get_text(strip=True)

Если разбор тяжёлый, вынесите его в пул процессов, чтобы не блокировать цикл:

import asyncio
from concurrent.futures import ProcessPoolExecutor

def heavy_parse(html):
    soup = BeautifulSoup(html, "lxml")
    return [a["href"] for a in soup.select("a")]

async def fetch_and_parse(session, url, pool):
    async with session.get(url) as resp:
        html = await resp.text()
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(pool, heavy_parse, html)

Подробно о парсерах — в хабе и в статье про lxml (самый быстрый вариант для async-нагрузок).

Часто асинхронно дёргают не HTML-страницы, а API — тогда тело ответа уже структурировано, и вместо парсера достаточно await resp.json(). Это и быстрее, и надёжнее разбора разметки; приёмы работы с такими ответами собраны в «Парсинг JSON на Python».

4. Кириллица в асинхронном парсинге

aiohttp при await resp.text() пытается определить кодировку по заголовкам. На рунет-сайтах это даёт сбои. Решения те же, что в синхронном коде:

# вариант 1: явная кодировка
html = await resp.text(encoding="utf-8")

# вариант 2: работать с байтами и отдать их парсеру
raw = await resp.read()
soup = BeautifulSoup(raw, "lxml")    # парсер сам прочитает <meta charset>

# вариант 3: ручное декодирование
html = raw.decode("windows-1251", errors="replace")

Полная теория проблемы — в хабе, раздел «Кириллица».

5. Управление параллельностью: семафоры

Запустить 10 000 запросов разом — значит «положить» и сервер, и свою сеть, и получить бан. Параллельность ограничивают семафором:

import asyncio
import aiohttp

async def fetch(session, url, semaphore):
    async with semaphore:                      # не больше N одновременно
        async with session.get(url) as resp:
            return await resp.text()

async def main(urls, concurrency=20):
    semaphore = asyncio.Semaphore(concurrency)
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url, semaphore) for url in urls]
        return await asyncio.gather(*tasks, return_exceptions=True)

Semaphore(20) гарантирует не более 20 активных запросов одновременно. Это ваш главный рычаг «вежливости»: подбирайте значение так, чтобы не перегружать целевой сайт. Добавьте небольшие случайные паузы (await asyncio.sleep(random.uniform(0.1, 0.5))) для естественности.

6. Прокси

В aiohttp прокси передаётся параметром запроса:

async with session.get(url, proxy="http://user:pass@ip:port") as resp:
    html = await resp.text()

Ротация — просто выбор случайного прокси на каждый запрос:

import random

PROXIES = ["http://ip1:port", "http://ip2:port", "http://ip3:port"]

async def fetch(session, url):
    proxy = random.choice(PROXIES)
    async with session.get(url, proxy=proxy) as resp:
        return await resp.text()

Общая стратегия работы с прокси (типы, отбраковка мёртвых) — в хабе, раздел «Прокси».

7. Парсинг через TOR

aiohttp не поддерживает SOCKS напрямую — нужен пакет aiohttp-socks:

# pip install aiohttp-socks
import aiohttp
from aiohttp_socks import ProxyConnector

async def main(urls):
    connector = ProxyConnector.from_url("socks5://127.0.0.1:9050")
    async with aiohttp.ClientSession(connector=connector) as session:
        async with session.get("https://httpbin.org/ip") as resp:
            print(await resp.json())   # IP выходного узла TOR

Смена выходного узла через сигнал NEWNYM (библиотека stem) описана в хабе, раздел «TOR». Учтите: TOR медленный — при высокой параллельности он станет узким местом.

8. HTTPS/SSL

По умолчанию aiohttp проверяет сертификаты. Отключить (только для отладки) или подставить свой контекст:

import ssl

# отключение проверки — НЕ для продакшена
async with session.get(url, ssl=False) as resp:
    ...

# свой SSL-контекст
ctx = ssl.create_default_context(cafile="/path/to/ca.crt")
async with session.get(url, ssl=ctx) as resp:
    ...

Принципы безопасности соединения — в хабе, раздел «HTTPS/SSL».

9. Работа с cookie

ClientSession хранит cookie между запросами автоматически — как requests.Session:

async with aiohttp.ClientSession() as session:
    # логин — сервер ставит cookie сессии
    await session.post("https://example.com/login",
                       data={"user": "u", "pass": "p"})
    # последующие запросы уже авторизованы
    async with session.get("https://example.com/profile") as resp:
        html = await resp.text()

Передать cookie вручную можно через параметр cookies={...}. Подробнее — в хабе, раздел «Cookie».

10. Статус ответа и заголовки

async with session.get(url) as resp:
    print(resp.status)                       # 200, 404 ...
    print(resp.headers.get("Content-Type"))
    if resp.status == 429:
        wait = int(resp.headers.get("Retry-After", 60))
        await asyncio.sleep(wait)            # не блокирует другие задачи!
    resp.raise_for_status()

Ключевое преимущество: await asyncio.sleep() при обработке 429 усыпляет только эту корутину, а остальные продолжают работать. В синхронном коде time.sleep() заморозил бы всё. Логика статусов — в хабе.

11. Очереди: asyncio.Queue

Для краулинга «по мере обнаружения ссылок» используют asyncio.Queue и пул воркеров-корутин:

import asyncio
import aiohttp

async def worker(name, queue, session, visited):
    while True:
        url = await queue.get()
        if url not in visited:
            visited.add(url)
            try:
                async with session.get(url) as resp:
                    html = await resp.text()
                # ... найти новые ссылки и положить в очередь:
                # for link in extract_links(html):
                #     await queue.put(link)
            except Exception as exc:
                print(f"{name} ошибка {url}: {exc}")
        queue.task_done()

async def crawl(start_urls, num_workers=10):
    queue = asyncio.Queue()
    visited = set()
    for url in start_urls:
        queue.put_nowait(url)

    async with aiohttp.ClientSession() as session:
        workers = [asyncio.create_task(worker(f"w{i}", queue, session, visited))
                   for i in range(num_workers)]
        await queue.join()          # ждём, пока очередь опустеет
        for w in workers:
            w.cancel()

set для дедупликации, Queue для координации воркеров — асинхронный аналог frontier из хаба. Для распределённого обхода очередь выносят в Redis. Промышленную реализацию этой схемы предоставляет Scrapy (он внутри тоже асинхронный).

12. httpx как альтернатива

httpx — современный клиент с одинаковым API для синхронного и асинхронного кода и поддержкой HTTP/2:

import httpx
import asyncio

async def main(urls):
    async with httpx.AsyncClient(http2=True, timeout=15) as client:
        tasks = [client.get(url) for url in urls]
        responses = await asyncio.gather(*tasks, return_exceptions=True)
        return responses

Если хочется один и тот же код переключать между sync/async и нужен HTTP/2 — httpx удобнее aiohttp. По чистой скорости на больших объёмах они сопоставимы.

13. Плюсы и минусы асинхронного парсинга

Плюсы:

• Огромная параллельность при минимуме памяти.
• Кратное ускорение на I/O-bound задачах.
• Дешёвые «паузы»: asyncio.sleep не блокирует другие задачи.
• Тонкий контроль скорости через семафоры.

Минусы:

• Сложнее писать и отлаживать (async/await повсюду).
• CPU-bound парсинг всё равно блокирует цикл — нужен пул процессов.
• Нельзя смешивать с блокирующими библиотеками без run_in_executor.
• Легко перегрузить целевой сайт — нужна дисциплина с семафорами.

Когда выбрать: тысячи+ страниц и важна скорость. Для пары сотен страниц проще requests + ThreadPoolExecutor. Для полноценного краулинга всего сайта — Scrapy, где асинхронность и очереди уже встроены. А если парсер живёт внутри веб-приложения, учтите, что ORM Django по-прежнему преимущественно синхронный и async-код там требует аккуратности (sync_to_async) — об этом в «Парсинг на Django».