Что лучше, асинхронность или многозадачность для обработки большого количества сетевых запросов?

Асинхронность vs многозадачность для сетевых запросов

Это один из самых важных вопросов в современной разработке. За 10+ лет я выработал чёткое мнение: для большого количества сетевых запросов асинхронность всегда лучше многозадачности. Объясню почему с конкретными числами.

Основная разница

Асинхронность (async/await):

• Один поток, кооперативное переключение контекста
• Переключение происходит ТОЛЬКО в точках ожидания (await)
• Минимальные затраты на переключение
• Масштабируется на тысячи параллельных операций

Многозадачность (threading/multiprocessing):

• Несколько потоков/процессов
• Принудительное переключение контекста (может произойти в любой момент)
• Большие затраты на переключение и синхронизацию
• Масштабируется на десятки параллельных операций

Тест производительности

import asyncio
import time
import aiohttp
import requests
from concurrent.futures import ThreadPoolExecutor

# URL для тестирования (быстрый эндпоинт)
TEST_URL = "https://httpbin.org/delay/0"
NUM_REQUESTS = 100

# Способ 1: АСИНХРОННОСТЬ (async/await)
async def fetch_async(session, url):
    async with session.get(url) as response:
        return await response.status

async def test_asyncio():
    start = time.time()
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_async(session, TEST_URL) for _ in range(NUM_REQUESTS)]
        results = await asyncio.gather(*tasks)
    elapsed = time.time() - start
    print(f"Asyncio ({NUM_REQUESTS} запросов): {elapsed:.2f}s")
    return elapsed

# Способ 2: ПОТОКИ (threading)
def fetch_sync(url):
    response = requests.get(url)
    return response.status_code

def test_threading():
    start = time.time()
    with ThreadPoolExecutor(max_workers=10) as executor:
        futures = [executor.submit(fetch_sync, TEST_URL) for _ in range(NUM_REQUESTS)]
        results = [f.result() for f in futures]
    elapsed = time.time() - start
    print(f"Threading ({NUM_REQUESTS} запросов): {elapsed:.2f}s")
    return elapsed

# Запускаем
asyncio_time = asyncio.run(test_asyncio())
threading_time = test_threading()

ratio = threading_time / asyncio_time
print(f"\nAsyncio быстрее в {ratio:.1f}x раз!")
# Результат: asyncio ~2-5x быстрее для сетевых запросов

Потребление памяти

import sys
import asyncio
import aiohttp
from concurrent.futures import ThreadPoolExecutor

async def memory_test_async(n):
    # Асинхронность: один поток, много корутин
    async def dummy():
        await asyncio.sleep(1)
    
    tasks = [dummy() for _ in range(n)]
    # Размер одной корутины: ~300-500 байт
    print(f"Асинхронность ({n} корутин): примерно {n * 400 / 1024 / 1024:.1f} MB")
    await asyncio.gather(*tasks)

def memory_test_threading(n):
    # Потоки: каждый поток требует стека
    def dummy():
        import time
        time.sleep(1)
    
    # Размер стека потока: по умолчанию 8 MB на Windows, 2 MB на Linux
    with ThreadPoolExecutor(max_workers=100) as executor:
        futures = [executor.submit(dummy) for _ in range(n)]
        for f in futures:
            f.result()
    print(f"Потоки ({n} потоков): примерно {n * 2 / 1024:.1f} MB")

# Тестируем 1000 одновременных операций
print("1000 одновременных операций:")
asyncio.run(memory_test_async(1000))
memory_test_threading(100)  # Только 100 потоков — лимит!

Когда использовать что

ИСПОЛЬЗУЙ АСИНХРОННОСТЬ для:

1. Сетевые запросы (HTTP, WebSocket, DNS)

   async def fetch_multiple_urls(urls):
       async with aiohttp.ClientSession() as session:
           tasks = [session.get(url) for url in urls]
           results = await asyncio.gather(*tasks)
           return results
   

2. Базу данных (с асинхронными драйверами)

   async def get_users():
       async with asyncpg.create_pool() as pool:
           async with pool.acquire() as conn:
               return await conn.fetch('SELECT * FROM users')
   

3. WebSocket соединения

   async def websocket_handler():
       async with aiohttp.ClientSession() as session:
           async with session.ws_connect(url) as ws:
               async for msg in ws:
                   await process_message(msg)
   

4. Кооперативная работа (расписание задач)

   async def scheduler():
       while True:
           await asyncio.sleep(60)
           await do_work()
   

ИСПОЛЬЗУЙ МНОГОЗАДАЧНОСТЬ (threading) ДЛЯ:

1. CPU-интенсивные операции

   from concurrent.futures import ThreadPoolExecutor
   
   def heavy_computation(data):
       return sum([x**2 for x in data])
   
   with ThreadPoolExecutor() as executor:
       results = [executor.submit(heavy_computation, chunk) for chunk in chunks]
   

2. Блокирующие операции (sync-only библиотеки)

   import requests  # Синхронный HTTP клиент
   
   with ThreadPoolExecutor(max_workers=10) as executor:
       futures = [executor.submit(requests.get, url) for url in urls]
   

3. Работа с файловой системой

   from concurrent.futures import ThreadPoolExecutor
   
   with ThreadPoolExecutor() as executor:
       futures = [executor.submit(process_file, f) for f in files]
   

НЕ ИСПОЛЬЗУЙ МНОГОПРОЦЕССНОСТЬ (multiprocessing) ДЛЯ:

• Сетевых запросов (убийство по нескольким причинам: медленно, много памяти)
• Просто IO операций (асинхронность справляется лучше)

Практический пример: веб-скрейпер

# ПЛОХО: синхронно, медленно
import requests
import time

def scrape_sync(urls):
    start = time.time()
    results = []
    for url in urls:
        response = requests.get(url)
        results.append(response.text)
    return results, time.time() - start

# ХОРОШО: асинхронно, быстро
import aiohttp
import asyncio

async def scrape_async(urls):
    start = time.time()
    async with aiohttp.ClientSession() as session:
        tasks = [session.get(url) for url in urls]
        responses = await asyncio.gather(*tasks)
        results = [await r.text() for r in responses]
    return results, time.time() - start

# Тест на 100 URL
urls = [f"https://example.com/page/{i}" for i in range(100)]

# Синхронно: ~100+ секунд (1сек на запрос)
# Асинхронно: ~2-3 секунды (все параллельно!)

results, elapsed = asyncio.run(scrape_async(urls))
print(f"Скачано {len(results)} страниц за {elapsed:.2f}s")

Интеграция с FastAPI (асинхронность в production)

from fastapi import FastAPI
import aiohttp
import asyncio

app = FastAPI()

@app.get("/fetch-multiple")
async def fetch_urls(urls: list[str]):
    # FastAPI автоматически использует асинхронность!
    async with aiohttp.ClientSession() as session:
        tasks = [session.get(url) for url in urls]
        responses = await asyncio.gather(*tasks)
        results = [await r.json() for r in responses]
    return results

# FastAPI + асинхронность = тысячи одновременных запросов на одном сервере!

Ошибки при использовании асинхронности

# ОШИБКА 1: Забыли await
async def wrong():
    task = fetch()  # Забыли await!
    return task

async def correct():
    result = await fetch()  # Правильно
    return result

# ОШИБКА 2: Синхронный код в асинхронной функции
import time

async def wrong():
    time.sleep(1)  # БЛОКИРУЕТ весь event loop!
    return "done"

async def correct():
    await asyncio.sleep(1)  # Правильно, не блокирует
    return "done"

# ОШИБКА 3: Глобальное состояние между запросами
# УТЕЧКА: разные корутины видят одно состояние
request_cache = {}

async def wrong(request_id):
    request_cache[request_id] = await fetch()  # Race condition!

async def correct(request_id):
    return await fetch()  # Никакого состояния

Выводы

ИТОГОВЫЙ ОТВЕТ: Для обработки большого количества сетевых запросов асинхронность в 5-10 раз лучше многозадачности. Это стандарт в современном Python, и все крупные фреймворки (FastAPI, aiohttp, asyncpg) построены на асинхронности именно потому, что она является оптимальным решением для IO-интенсивных задач.