Как работает Asyncio?

Как работает Asyncio?

Asyncio — это встроенная библиотека Python для асинхронного программирования. Она позволяет писать многозадачный код с одним потоком, эффективно используя ввод-вывод.

Основной принцип

Традиционное программирование (синхронное):
┌─────────────────────────────┐
│ 1. Запрос к БД (1 сек)      │ ← БЛОКИРУЕТ
│ 2. Обработка (0.1 сек)      │
│ 3. Запрос к API (2 сек)     │ ← БЛОКИРУЕТ
│ 4. Ответ (0.1 сек)          │
├─────────────────────────────┤
Текущее время: 3.2 секунды
└─────────────────────────────┘

Асинхронное программирование (asyncio):
┌─────────────────────────────────┐
│ 1. Запрос к БД (1 сек)      ┐   │
│ 3. Запрос к API (2 сек)     ├─ ПАРАЛЛЕЛЬНО!
│ 2. Обработка (0.1 сек)      ┘   │
│ 4. Ответ (0.1 сек)              │
├─────────────────────────────────┤
Текущее время: 2.2 секунды (1 + 2 пока обрабатываем)
└─────────────────────────────────┘

Event Loop — сердце asyncio

import asyncio

# Event loop — это бесконечный цикл, который запускает корутины

async def task1():
    print("Task 1 started")
    await asyncio.sleep(1)  # симуляция блокирующей операции
    print("Task 1 finished")

async def task2():
    print("Task 2 started")
    await asyncio.sleep(2)
    print("Task 2 finished")

async def main():
    # Запускаем обе задачи одновременно
    await asyncio.gather(task1(), task2())
    # Работает примерно 2 секунды (не 3!)

asyncio.run(main())

# Event loop запускает main()
# Видит await asyncio.sleep(1) в task1 — паузирует task1
# Видит await asyncio.sleep(2) в task2 — паузирует task2
# Ждёт 1 сек → task1 продолжается
# Ждёт 2 сек → task2 продолжается
# Всё выполнилось за 2 сек!

Корутины (Coroutines)

Корутина — это функция, которая может приостанавливаться и возобновляться.

# Синтаксис: async def

async def fetch_data():
    print("Fetching...")
    await asyncio.sleep(2)  # ждём без блокирования
    return "Data"

# Создание корутины
coro = fetch_data()
print(coro)  # <coroutine object fetch_data at ...>

# Запуск корутины
result = asyncio.run(fetch_data())
print(result)  # "Data"

# ВНЕ asyncio.run() нельзя использовать await!
# await fetch_data()  # RuntimeError: no running event loop

Await — точка паузы

import asyncio
import time

async def slow_function():
    print(f"{time.time()}: Starting")
    await asyncio.sleep(2)  # Здесь корутина паузируется
    print(f"{time.time()}: Done")
    return "Result"

async def main():
    # Запусти и жди результат
    result = await slow_function()
    print(result)

asyncio.run(main())

# Вывод:
# 1234567890.1: Starting
# [1234567892.1]: Done  # через 2 секунды
# Result

Важное: разница между async функциями и обычными

import asyncio

# ✅ Асинхронная функция
async def async_func():
    await asyncio.sleep(1)
    return "async"

# ❌ Обычная функция
def sync_func():
    time.sleep(1)  # БЛОКИРУЕТ!
    return "sync"

async def main():
    # await работает только с async функциями
    result = await async_func()  # OK
    
    # await с обычной функцией — бессмыслица
    result = await sync_func()  # TypeError!
    
    # Обычную функцию просто вызываешь
    result = sync_func()  # OK, но блокирует event loop!

Запуск множества корутин

import asyncio
import aiohttp

async def fetch_url(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    urls = [
        "https://example.com",
        "https://python.org",
        "https://github.com"
    ]
    
    async with aiohttp.ClientSession() as session:
        # 1. ПОСЛЕДОВАТЕЛЬНО (МЕДЛЕННО)
        for url in urls:
            await fetch_url(session, url)
        # Занимает: 3 сек + 3 сек + 3 сек = 9 сек
        
        # 2. ПАРАЛЛЕЛЬНО (БЫСТРО)
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        # Занимает: max(3, 3, 3) = 3 сек

asyncio.run(main())

asyncio.gather() vs asyncio.create_task()

import asyncio

async def task():
    await asyncio.sleep(1)
    return "result"

# 1. gather — запустить и ждать
async def using_gather():
    results = await asyncio.gather(task(), task(), task())
    # Запускает все и ждёт все
    return results

# 2. create_task — запустить в фоне
async def using_create_task():
    t1 = asyncio.create_task(task())
    t2 = asyncio.create_task(task())
    
    # Они уже выполняются!
    
    result1 = await t1
    result2 = await t2
    # Ждём результаты

# 3. TaskGroup (Python 3.11+) — структурированная конкурентность
async def using_task_group():
    async with asyncio.TaskGroup() as tg:
        t1 = tg.create_task(task())
        t2 = tg.create_task(task())
        # Автоматически ждёт все при выходе

Таймауты и обработка ошибок

import asyncio

async def slow_task():
    await asyncio.sleep(10)

async def main():
    # 1. TIMEOUT
    try:
        await asyncio.wait_for(slow_task(), timeout=2)
    except asyncio.TimeoutError:
        print("Task timed out!")
    
    # 2. ОБРАБОТКА ОШИБОК В gather
    results = await asyncio.gather(
        slow_task(),
        asyncio.sleep(0),
        slow_task(),
        return_exceptions=True  # не прерывать при ошибке
    )
    # results = [TimeoutError, None, TimeoutError]
    
    # 3. ОБРАБОТКА ОШИБОК В TaskGroup (3.11+)
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(slow_task())
            tg.create_task(asyncio.sleep(0))
    except* asyncio.TimeoutError:  # except* для группы ошибок
        print("Some tasks timed out")

asyncio.run(main())

Реальный пример: веб-скрейпер

import asyncio
import aiohttp
from typing import List

class WebScraper:
    def __init__(self, max_concurrent=10):
        self.semaphore = asyncio.Semaphore(max_concurrent)
    
    async def fetch(self, session: aiohttp.ClientSession, url: str) -> str:
        # Ограничиваем количество одновременных запросов
        async with self.semaphore:
            try:
                async with session.get(url, timeout=5) as response:
                    return await response.text()
            except Exception as e:
                print(f"Error fetching {url}: {e}")
                return None
    
    async def scrape_many(self, urls: List[str]) -> List[str]:
        async with aiohttp.ClientSession() as session:
            tasks = [self.fetch(session, url) for url in urls]
            return await asyncio.gather(*tasks)

# Использование
urls = [f"https://example.com/page/{i}" for i in range(100)]
scraper = WebScraper(max_concurrent=10)
results = asyncio.run(scraper.scrape_many(urls))
print(f"Fetched {len(results)} pages")

Сравнение: Threads vs Asyncio vs Multiprocessing

import asyncio
import threading
import multiprocessing
import time

def io_task():  # I/O — блокирующее
    requests.get("https://httpbin.org/delay/1")

def cpu_task():  # CPU — вычисления
    return sum(i**2 for i in range(100_000_000))

# I/O-BOUND: Asyncio ЛУЧШЕ
# Threading: ~2s (10 потоков на 10 задачах)
# Asyncio: ~1s (1 поток, 10 задач параллельно)
# Multiprocessing: ~2-3s (overhead на создание)

# CPU-BOUND: Multiprocessing ЛУЧШЕ
# Threading: ~10s (GIL!)
# Asyncio: ~10s (одно вычисление за раз)
# Multiprocessing: ~3s (на 4-ядерном процессоре)

Антипаттерны

# ❌ Смешивание sync и async
async def bad():
    time.sleep(1)  # БЛОКИРУЕТ event loop!
    await asyncio.sleep(1)

# ✅ Правильно
async def good():
    await asyncio.sleep(1)  # не блокирует

# ❌ Забыли await
async def bad2():
    coro = asyncio.sleep(1)  # создали, но не запустили!
    print("Done")  # выполнится сразу

# ✅ Правильно
async def good2():
    await asyncio.sleep(1)  # запустили и ждём

# ❌ Долгие операции в event loop
async def bad3():
    data = expensive_calculation()  # 10 сек, БЛОКИРУЕТ!
    await asyncio.sleep(1)

# ✅ Правильно
async def good3():
    # Запустить в отдельном потоке
    loop = asyncio.get_event_loop()
    data = await loop.run_in_executor(None, expensive_calculation)
    await asyncio.sleep(1)

Резюме

Asyncio работает на базе event loop (событийного цикла), который запускает многие корутины в одном потоке:

1. Event Loop — бесконечный цикл, управляющий корутинами
2. Coroutines — функции async def, которые можно паузировать
3. await — точка, где корутина паузируется и отпускает event loop
4. Tasks — запущенные корутины с которыми можно работать

Идеально для:

• I/O-bound приложений (веб-сервисы, API клиенты)
• Обработки множества одновременных соединений
• Сетевых операций и файловых операций

Не подходит для:

• CPU-bound вычислений (используй multiprocessing)
• Синхронного кода (смешивать нельзя)
• Если нужна простота (threading проще для I/O)