Что такое потоки, процессы и асинхронность в Python, и чем они отличаются?

Потоки, процессы и асинхронность в Python

Это три разных подхода к параллельному выполнению кода в Python. Каждый имеет свои преимущества и недостатки, и выбор зависит от типа задачи.

GIL (Global Interpreter Lock) — ключ ко всему

Прежде чем разбираться в различиях, нужно понять GIL. Это мьютекс, который позволяет только одному потоку исполнять Python код одновременно. Это значит, что несколько потоков не дают реального параллелизма для CPU-bound задач.

1. Потоки (Threads) — для I/O-bound задач

Потоки делят один процесс и одно адресное пространство памяти. Они легкие и быстрые для создания, но заблокированы GIL-ом для вычислений.

import threading
import time
import requests

def fetch_data(url):
    # I/O операция — потоки отпускают GIL при ожидании
    response = requests.get(url, timeout=5)
    return response.status_code

# Создание потоков
start = time.time()
threads = []
for i in range(5):
    thread = threading.Thread(target=fetch_data, args=(f"https://httpbin.org/delay/2",))
    threads.append(thread)
    thread.start()

# Ожидание завершения
for thread in threads:
    thread.join()

print(f"Время выполнения: {time.time() - start:.2f}c")  # ~2 сек (параллельно)

# Потокобезопасность
lock = threading.Lock()
counter = 0

def increment():
    global counter
    with lock:  # Синхронизация
        counter += 1

threads = [threading.Thread(target=increment) for _ in range(100)]
for t in threads:
    t.start()
for t in threads:
    t.join()

print(counter)  # 100 (безопасно)

2. Процессы (Processes) — для CPU-bound задач

Процессы имеют отдельный интерпретатор Python, отдельный GIL и отдельное адресное пространство. Они дают настоящий параллелизм для вычислений, но тяжелее в создании и использовании памяти.

import multiprocessing
import time

def cpu_bound_task(n):
    # CPU-bound операция — потоки здесь неэффективны
    total = sum(range(n))
    return total

if __name__ == "__main__":
    # Важно: используй if __name__ == "__main__" на Windows/macOS
    
    # Последовательное выполнение
    start = time.time()
    for i in range(4):
        cpu_bound_task(100000000)
    print(f"Последовательно: {time.time() - start:.2f}c")  # ~8 сек
    
    # Параллельное с процессами
    start = time.time()
    with multiprocessing.Pool(processes=4) as pool:
        results = pool.map(cpu_bound_task, [100000000] * 4)
    print(f"С процессами: {time.time() - start:.2f}c")  # ~2 сек (на 4-ядерном)
    
    # Передача данных между процессами
    queue = multiprocessing.Queue()
    
    def producer(q):
        for i in range(10):
            q.put(i)
    
    def consumer(q):
        while True:
            try:
                value = q.get(timeout=1)
                print(f"Получено: {value}")
            except:
                break
    
    p1 = multiprocessing.Process(target=producer, args=(queue,))
    p2 = multiprocessing.Process(target=consumer, args=(queue,))
    
    p1.start()
    p2.start()
    
    p1.join()
    p2.terminate()

3. Асинхронность (Async/Await) — для I/O-bound задач (эффективнее потоков)

Асинхронное программирование — это один поток, который переключается между задачами во время I/O операций. Это самое эффективное решение для I/O-bound задач.

import asyncio
import aiohttp
import time

async def fetch_data(session, url):
    async with session.get(url) as response:
        return response.status

async def main():
    start = time.time()
    
    # Создание сессии для переиспользования соединений
    async with aiohttp.ClientSession() as session:
        # Создаём список задач
        tasks = [
            fetch_data(session, "https://httpbin.org/delay/2")
            for _ in range(5)
        ]
        
        # Выполняем все задачи параллельно
        results = await asyncio.gather(*tasks)
    
    print(f"Результаты: {results}")
    print(f"Время выполнения: {time.time() - start:.2f}c")  # ~2 сек

# asyncio.run(main())  # Python 3.7+

# Обработка исключений в asyncio
async def handle_error():
    try:
        await asyncio.sleep(1)
        raise ValueError("Ошибка!")
    except ValueError as e:
        print(f"Поймали ошибку: {e}")

asyncio.run(handle_error())

# Создание очереди для асинхронных задач
async def producer(queue):
    for i in range(5):
        await queue.put(i)
        print(f"Произведено: {i}")

async def consumer(queue):
    while True:
        value = await queue.get()
        if value is None:
            break
        print(f"Потреблено: {value}")

async def async_main():
    queue = asyncio.Queue()
    
    # Запускаем производителя и потребителя параллельно
    await asyncio.gather(
        producer(queue),
        consumer(queue)
    )

# asyncio.run(async_main())

Сравнительная таблица

┌─────────────────┬──────────────┬──────────────┬──────────────┐
│ Параметр        │ Потоки       │ Процессы     │ Асинхронность│
├─────────────────┼──────────────┼──────────────┼──────────────┤
│ GIL             │ Да (блокирует)│ Нет (каждый) │ Не применяется│
│ CPU-bound       │ Плохо        │ Хорошо       │ Плохо        │
│ I/O-bound       │ Хорошо       │ Тяжело       │ Отлично      │
│ Память          │ Мало         │ Много        │ Очень мало   │
│ Простота        │ Средняя      │ Средняя      │ Сложная      │
│ Параллелизм     │ Нет (один CPU)│ Да (несколько)│ Нет (один CPU)│
│ Масштаб         │ Сотни        │ Десятки      │ Тысячи       │
└─────────────────┴──────────────┴──────────────┴──────────────┘

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

# 1. I/O-bound задачи (веб-запросы, БД запросы) → Асинхронность
async def web_scraper():
    async with aiohttp.ClientSession() as session:
        # Отличная масштабируемость
        pass

# 2. CPU-bound задачи (вычисления, обработка) → Процессы
with multiprocessing.Pool(4) as pool:
    # Настоящий параллелизм
    results = pool.map(expensive_calculation, data)

# 3. Простые I/O задачи → Потоки
threads = [threading.Thread(target=simple_io_task) for _ in range(10)]
# Начиная с тысяч задач → переходи на асинхронность

# 4. Смешанные нагрузки
with multiprocessing.Pool(4) as pool:
    # Каждый процесс может использовать асинхронность
    results = pool.starmap(async_wrapper, tasks)

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

import asyncio
import aiohttp

async def fetch_page(session, url):
    try:
        async with session.get(url, timeout=5) as response:
            return await response.text()
    except Exception as e:
        print(f"Ошибка при загрузке {url}: {e}")
        return None

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

urls = ["https://httpbin.org/delay/1" for _ in range(10)]
# asyncio.run(scrape_urls(urls))  # Загрузит 10 страниц за ~1 сек

Резюме

• Потоки (threading): Хороши для I/O-bound задач, ограничены GIL, просто использовать
• Процессы (multiprocessing): Необходимы для CPU-bound задач, true параллелизм, больше памяти
• Асинхронность (asyncio): Лучший выбор для I/O-bound, тысячи одновременных задач, требует переписи кода
• Выбор: CPU → multiprocessing, I/O много → asyncio, I/O мало → threading