Асинхронный код исполняется в одном потоке или нет

# Асинхронный код в Python: потоки и выполнение

Краткий ответ

Асинхронный код в Python исполняется в одном потоке. Это достигается благодаря event loop'у, который переключается между асинхронными задачами (coroutines) во время их блокировок (I/O операций).

Понимание асинхронности

Асинхронность в Python — это не параллелизм, а concurrent (совместное) выполнение. Это разные вещи:

Параллелизм (Parallelism)

import threading

# Две функции выполняются в разных потоках ОДНОВРЕМЕННО
def task1():
    print("Задача 1 работает на потоке", threading.current_thread().name)
    time.sleep(1)
    print("Задача 1 завершена")

def task2():
    print("Задача 2 работает на потоке", threading.current_thread().name)
    time.sleep(1)
    print("Задача 2 завершена")

thread1 = threading.Thread(target=task1)
thread2 = threading.Thread(target=task2)

thread1.start()  # Разные потоки
thread2.start()  # Выполняются параллельно на разных ядрах CPU
thread1.join()
thread2.join()

Асинхронность (Concurrency)

import asyncio

# Обе корутины выполняются в одном потоке ПООЧЕРЕДНО
async def task1():
    print("Задача 1 начата")
    await asyncio.sleep(1)  # Освобождает поток
    print("Задача 1 завершена")

async def task2():
    print("Задача 2 начата")
    await asyncio.sleep(1)  # Освобождает поток
    print("Задача 2 завершена")

async def main():
    # Обе задачи работают в одном потоке
    await asyncio.gather(task1(), task2())

asyncio.run(main())
# Выведет:
# Задача 1 начата
# Задача 2 начата
# Задача 1 завершена
# Задача 2 завершена

Event Loop в Python

Event loop — это сердце асинхронности. Это цикл, запущенный в одном потоке, который управляет выполнением корутин:

import asyncio
import threading

async def print_info():
    loop = asyncio.get_event_loop()
    print(f"Event loop работает в потоке: {threading.current_thread().name}")
    print(f"ID потока: {threading.get_ident()}")
    await asyncio.sleep(0)
    print(f"ID потока после await: {threading.get_ident()}")

asyncio.run(print_info())
# Выведет:
# Event loop работает в потоке: MainThread
# ID потока: 140234567890
# ID потока после await: 140234567890 (один и тот же!)

Как работает event loop

Event loop выполняет следующий алгоритм:

1. Получить следующую задачу из очереди
2. Выполнить её до первого await (или до завершения)
3. Если задача дошла до await:
   - Поместить задачу в очередь ожидания
   - Перейти к следующей задаче
4. Когда I/O операция завершилась:
   - Вернуть управление задаче
   - Продолжить выполнение
5. Повторять пока все задачи не завершены

Визуализация

import asyncio
import time

async def task(name, duration):
    print(f"[{time.time():.2f}] {name} начата")
    await asyncio.sleep(duration)
    print(f"[{time.time():.2f}] {name} завершена")

async def main():
    # Все задачи работают в одном потоке
    await asyncio.gather(
        task("Задача 1", 2),
        task("Задача 2", 1),
        task("Задача 3", 3),
    )

start = time.time()
asyncio.run(main())
print(f"Всего времени: {time.time() - start:.2f}s")

# Выведет (общее время ~3 сек, а не 6!):
# [1705000000.00] Задача 1 начата
# [1705000000.00] Задача 2 начата
# [1705000000.00] Задача 3 начата
# [1705000001.00] Задача 2 завершена
# [1705000002.00] Задача 1 завершена
# [1705000003.00] Задача 3 завершена
# Всего времени: 3.00s

GIL и асинхронность

Python имеет GIL (Global Interpreter Lock), который предотвращает выполнение нескольких потоков с Python кодом одновременно. Асинхронность НЕ избегает GIL, потому что всё работает в одном потоке:

import asyncio
import threading

async def cpu_bound_task(n):
    """Задача с высокой нагрузкой на CPU"""
    result = 0
    for i in range(n):
        result += i
    return result

async def main():
    # Обе задачи работают в одном потоке
    # Если одна из них блокирует CPU, вторая не сможет выполняться!
    results = await asyncio.gather(
        cpu_bound_task(10000000),
        cpu_bound_task(10000000),
    )
    print(results)

asyncio.run(main())
# Выполняются ПОСЛЕДОВАТЕЛЬНО (не параллельно)
# потому что обе требуют CPU и работают в одном потоке

I/O bound vs CPU bound

I/O bound (идеально для async)

import asyncio
import aiohttp

async def fetch_url(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            print(f"Получена ответ: {response.status}")
            return await response.text()

async def main():
    urls = [
        "https://example.com",
        "https://google.com",
        "https://github.com",
    ]
    # Все запросы выполняются в одном потоке, но параллельно!
    results = await asyncio.gather(*[fetch_url(url) for url in urls])
    print(f"Получено {len(results)} ответов")

asyncio.run(main())
# Все 3 запроса выполняются одновременно в одном потоке

CPU bound (НЕ для async, используй ProcessPool)

import asyncio
from concurrent.futures import ProcessPoolExecutor

def cpu_heavy_calculation(n):
    """Тяжелый расчёт"""
    result = 0
    for i in range(n):
        result += i ** 2
    return result

async def main():
    # Для CPU-bound работы используй ProcessPoolExecutor
    loop = asyncio.get_event_loop()
    with ProcessPoolExecutor() as executor:
        result1 = await loop.run_in_executor(executor, cpu_heavy_calculation, 10000000)
        result2 = await loop.run_in_executor(executor, cpu_heavy_calculation, 10000000)
    print(f"Результаты: {result1}, {result2}")

asyncio.run(main())
# Теперь два процесса работают на разных ядрах CPU параллельно

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

import asyncio
import aiohttp
from time import time

async def fetch_page(session, url):
    try:
        async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
            return await resp.text()
    except Exception as e:
        return f"Ошибка: {e}"

async def scrape_multiple_sites(urls):
    # Все запросы в одном потоке, но параллельно
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_page(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        return results

async def main():
    urls = [
        "https://example.com",
        "https://google.com",
        "https://github.com",
        "https://stackoverflow.com",
    ]
    
    start = time()
    results = await scrape_multiple_sites(urls)
    elapsed = time() - start
    
    print(f"Получено {len(results)} страниц за {elapsed:.2f} сек")
    for i, content in enumerate(results, 1):
        print(f"URL {i}: {len(content)} символов")

asyncio.run(main())
# Все 4 запроса выполняются параллельно в одном потоке
# Общее время ~ самому долгому запросу (~2-3 сек)
# А не сумме времён (~ 8-12 сек)

Как проверить количество потоков

import asyncio
import threading

async def check_threads():
    print(f"Активных потоков: {threading.active_count()}")
    print(f"Текущий поток: {threading.current_thread().name}")
    
    # После await
    await asyncio.sleep(0)
    
    print(f"Активных потоков после await: {threading.active_count()}")
    print(f"Текущий поток после await: {threading.current_thread().name}")

asyncio.run(check_threads())
# Выведет:
# Активных потоков: 1
# Текущий поток: MainThread
# Активных потоков после await: 1
# Текущий поток после await: MainThread

Сравнение подходов

Выводы

✅ Асинхронный код в Python работает в одном потоке ✅ Event loop переключается между задачами при блокировках (I/O) ✅ Это дешевле чем потоки (нет overhead создания потока) ✅ Идеально для I/O-bound задач (сеть, файлы, БД) ✅ НЕ подходит для CPU-bound задач (используй multiprocessing) ✅ GIL не помогает асинхронности, так как всё в одном потоке ✅ Всегда помещай I/O операции в await для максимальной効率