Какие знаешь блокирующие функции при работе с асинхронностью?

Какие знаешь блокирующие функции при работе с асинхронностью?

Это один из самых важных вопросов для async разработчика. Блокирующие функции в async коде — это главный источник проблем с производительностью. Расскажу, какие функции блокируют и как их избежать.

Что такое блокирующая функция?

Это функция, которая останавливает весь event loop, пока она выполняется. В это время другие корутины не могут работать.

import asyncio
import time

# ПЛОХО: блокирующая функция в асинхронном коде
async def bad_async_func():
    print("Start")
    time.sleep(5)  # БЛОКИРУЕТ весь event loop на 5 секунд!
    print("End")

# ХОРОШО: асинхронная альтернатива
async def good_async_func():
    print("Start")
    await asyncio.sleep(5)  # Не блокирует, другие корутины работают
    print("End")

Список блокирующих функций

1. time.sleep() — КЛАССИЧЕСКАЯ ОШИБКА

# ПЛОХО
async def bad_delay():
    print("Start:", time.time())
    time.sleep(3)  # Весь event loop остановлен!
    print("End:", time.time())

# ПРАВИЛЬНО
async def good_delay():
    print("Start:", time.time())
    await asyncio.sleep(3)  # Только эта корутина ждёт
    print("End:", time.time())

# Демонстрация
async def main():
    task1 = good_delay()
    task2 = good_delay()
    
    start = time.time()
    await asyncio.gather(task1, task2)
    print(f"Время: {time.time() - start:.1f}s")  # 3s (параллельно!)
    # Если бы использовали time.sleep(), было бы 6s

2. Файловый ввод-вывод (I/O операции)

# ПЛОХО
async def bad_file_read():
    with open('large_file.txt', 'r') as f:  # БЛОКИРУЕТ!
        data = f.read()
    return data

# ПРАВИЛЬНО (используй aiofiles)
import aiofiles

async def good_file_read():
    async with aiofiles.open('large_file.txt', 'r') as f:
        data = await f.read()  # Не блокирует
    return data

3. Сетевые операции

# ПЛОХО (используешь requests)
import requests

async def bad_http_request():
    response = requests.get('https://api.example.com/data')  # БЛОКИРУЕТ!
    return response.json()

# ПРАВИЛЬНО (используй aiohttp или httpx)
import aiohttp

async def good_http_request():
    async with aiohttp.ClientSession() as session:
        async with session.get('https://api.example.com/data') as response:
            return await response.json()  # Не блокирует

# Или httpx
import httpx

async def good_http_request_v2():
    async with httpx.AsyncClient() as client:
        response = await client.get('https://api.example.com/data')
        return response.json()

4. Операции с базой данных

# ПЛОХО (синхронные драйверы)
from sqlalchemy import create_engine

async def bad_db_query():
    engine = create_engine('postgresql://...')
    with engine.connect() as conn:  # БЛОКИРУЕТ!
        result = conn.execute("SELECT * FROM users")
    return result.fetchall()

# ПРАВИЛЬНО (асинхронные драйверы)
import databases  # или asyncpg, aiomysql, motor для MongoDB

database = databases.Database('postgresql://...')

async def good_db_query():
    await database.connect()
    rows = await database.fetch('SELECT * FROM users')  # Не блокирует
    return rows

# Или SQLAlchemy 2.0+ с asyncio
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession

async def good_db_query_v2():
    engine = create_async_engine('postgresql+asyncpg://...')
    async with AsyncSession(engine) as session:
        result = await session.execute('SELECT * FROM users')
        return result.fetchall()

5. Обработка JSON и парсинг

# ПЛОХО
import json

async def bad_json_parse():
    large_json = json.dumps({...})  # Может быть медленно
    data = json.loads(large_json)  # БЛОКИРУЕТ для больших файлов
    return data

# ПРАВИЛЬНО
import orjson  # Быстрый JSON парсер

async def good_json_parse():
    large_json = "{...}"
    # orjson быстрее, или используй loop.run_in_executor()
    data = orjson.loads(large_json)
    return data

6. Интенсивные вычисления (CPU-bound операции)

# ПЛОХО
async def bad_computation():
    result = sum(i**2 for i in range(100_000_000))  # БЛОКИРУЕТ!
    return result

# ПРАВИЛЬНО (используй run_in_executor)
import asyncio
from concurrent.futures import ProcessPoolExecutor

async def good_computation():
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(
        ProcessPoolExecutor(),
        lambda: sum(i**2 for i in range(100_000_000))
    )
    return result

7. Модули, которые блокируют

# Блокирующие модули
requests          # HTTP запросы
sqlalchemy        # Синхронный ORM
pyself.psycopg2   # Синхронный PostgreSQL драйвер
mysql.connector   # MySQL
mongo (синхронный)  # MongoDB синхронный
pickle            # Для больших объектов
import subprocess # Запуск процессов
os.system()       # Системные команды
time.sleep()      # Задержки

Правильные асинхронные альтернативы

# Таблица замен

Блокирующие         →  Асинхронные
=====================================
time.sleep()         →  asyncio.sleep()
requests.get()       →  aiohttp / httpx (async)
sqlalchemy (sync)    →  databases / sqlalchemy async
psycopg2             →  asyncpg
mysql.connector      →  aiomysql
open() file I/O      →  aiofiles
subprocess.run()     →  asyncio.create_subprocess_exec()
os.system()          →  asyncio.create_subprocess_shell()
json (большие)       →  orjson
CPU вычисления       →  loop.run_in_executor()

Практический пример: Проверка блокирования

import asyncio
import time

# ПЛОХОЙ КОД
async def slow_task():
    print(f"{time.time()}: Task 1 start")
    time.sleep(2)  # БЛОКИРУЕТ
    print(f"{time.time()}: Task 1 end")

async def slow_task_2():
    print(f"{time.time()}: Task 2 start")
    await asyncio.sleep(2)  # Не блокирует
    print(f"{time.time()}: Task 2 end")

async def main():
    # Запускаем две задачи параллельно
    await asyncio.gather(
        slow_task(),      # Блокирует весь event loop
        slow_task_2()     # Ждёт в очереди
    )

# Результат:
# 1234567890.1: Task 1 start
# 1234567892.1: Task 1 end      <- 2 СЕКУНДЫ!
# 1234567892.1: Task 2 start
# 1234567894.1: Task 2 end
# Общее время: 4 СЕКУНДЫ (последовательно вместо параллельно!)

# ПРАВИЛЬНЫЙ КОД
async def main_correct():
    await asyncio.gather(
        slow_task_2(),    # Не блокирует
        slow_task_2()     # Работает параллельно
    )

# Результат:
# 1234567890.1: Task 2 start
# 1234567890.1: Task 2 start
# 1234567892.1: Task 2 end
# 1234567892.1: Task 2 end
# Общее время: 2 СЕКУНДЫ (параллельно!)

Как найти блокирующий код?

# Инструмент для отладки
import asyncio

class SlowCallback:
    def __init__(self, threshold=0.1):
        self.threshold = threshold
    
    def __call__(self, future):
        if future._loop.time() - future._time > self.threshold:
            print(f"Slow callback: {future._time}")

# Или используй профайлер
import cProfile
import pstats

if __name__ == '__main__':
    profiler = cProfile.Profile()
    profiler.enable()
    asyncio.run(main())
    profiler.disable()
    stats = pstats.Stats(profiler)
    stats.sort_stats('cumulative')
    stats.print_stats(10)  # Топ 10 медленных операций

На собеседовании

"Блокирующие функции в async коде — это операции, которые остановят весь event loop.

Самые частые:

• time.sleep() → используй asyncio.sleep()
• requests → используй aiohttp или httpx
• Синхронные БД драйверы → используй asyncpg, aiomysql, motor
• Файловый I/O → используй aiofiles
• Интенсивные вычисления → используй loop.run_in_executor()

Главное правило: если используешь async, ВСЕ операции должны быть асинхронными. Иначе теряется весь смысл асинхронности.

Для поиска: включу await везде, где нужна асинхронность."