Многопоточность и асинхронность

Отличие потоков от процессов, GIL в Python, пул потоков и процессов, состояние гонки и блокировки, async/await и событийный цикл, выбор подхода под задачу.

Программа с параллельной обработкой — выигрышная тема для курсовой: есть что измерить и с чем сравнить. Но выбор между потоками, процессами и асинхронностью зависит от того, чем именно занята программа.

Два типа нагрузки

Различие принципиальное. Пока программа ждёт ответ сервера, процессор простаивает — и потоки позволяют занять это время другими запросами. Но если процессор загружен вычислениями, добавление потоков не ускорит работу: ядер-то не стало больше.

Потоки и процессы

GIL в Python

В CPython есть глобальная блокировка интерпретатора: в один момент байт-код исполняет только один поток. Поэтому вычислительные задачи потоками в Python не ускоряются — нужны процессы. На операциях ввода-вывода GIL освобождается, и там потоки работают полноценно.

import time
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

def cpu_task(n):
    """Загружает процессор."""
    return sum(i * i for i in range(n))

def io_task(url):
    """Ждёт сеть."""
    time.sleep(0.5)          # имитация запроса
    return url

DATA = [5_000_000] * 4

# Последовательно
t = time.perf_counter()
[cpu_task(n) for n in DATA]
print(f'последовательно: {time.perf_counter()-t:.2f} с')

# Потоки на вычислениях — выигрыша нет из-за GIL
t = time.perf_counter()
with ThreadPoolExecutor(max_workers=4) as ex:
    list(ex.map(cpu_task, DATA))
print(f'потоки:          {time.perf_counter()-t:.2f} с')

# Процессы — ускорение почти в число ядер
t = time.perf_counter()
with ProcessPoolExecutor(max_workers=4) as ex:
    list(ex.map(cpu_task, DATA))
print(f'процессы:        {time.perf_counter()-t:.2f} с')

Состояние гонки

import threading

counter = 0

def increment_unsafe(times):
    global counter
    for _ in range(times):
        counter += 1        # три операции: чтение, сложение, запись

threads = [threading.Thread(target=increment_unsafe, args=(100_000,)) for _ in range(4)]
for t in threads: t.start()
for t in threads: t.join()
print(counter)   # ожидаем 400 000, получаем меньше — часть увеличений потерялась

# С блокировкой результат верный
lock = threading.Lock()
counter = 0

def increment_safe(times):
    global counter
    for _ in range(times):
        with lock:
            counter += 1

Причина потерь: counter += 1 не атомарна. Поток читает значение, но до записи его вытесняет другой поток, который читает то же старое значение. Оба записывают одно и то же — одно увеличение пропало. Такие ошибки воспроизводятся не всегда, что делает их особенно неприятными.

Средства синхронизации

import threading, queue, time

tasks = queue.Queue(maxsize=10)
results = queue.Queue()

def worker(wid):
    while True:
        item = tasks.get()
        if item is None:            # сигнал завершения
            tasks.task_done()
            break
        results.put((wid, item * 2))
        tasks.task_done()

workers = [threading.Thread(target=worker, args=(i,), daemon=True) for i in range(4)]
for w in workers: w.start()

for i in range(20): tasks.put(i)
for _ in workers: tasks.put(None)
tasks.join()

Асинхронность

import asyncio, time

async def fetch(session_id, delay):
    print(f'запрос {session_id} начат')
    await asyncio.sleep(delay)          # отдаём управление на время ожидания
    print(f'запрос {session_id} готов')
    return session_id

async def main():
    t = time.perf_counter()
    # Все запросы идут одновременно в одном потоке
    results = await asyncio.gather(
        fetch(1, 1.0), fetch(2, 1.5), fetch(3, 0.5)
    )
    print(f'готово за {time.perf_counter()-t:.2f} с')   # ~1.5 с, не 3.0
    return results

asyncio.run(main())

Асинхронность работает в одном потоке: пока одна корутина ждёт, событийный цикл выполняет другие. Гонок за данные нет по построению — переключение происходит только на await, в предсказуемых точках.

Как выбрать

1. Программа считает — процессы, число рабочих по числу ядер.
2. Программа обращается к сети или диску, задач немного — потоки, они проще.
3. Много одновременных сетевых операций (сотни и тысячи) — async.
4. Смешанная нагрузка — async для ввода-вывода плюс пул процессов для расчётов через run_in_executor.
5. Задачи независимы и данных много — процессы: изоляция избавляет от синхронизации.

Что показать в курсовой

• Реализацию задачи в трёх вариантах: последовательно, потоками, процессами.
• Таблицу и график времени работы от числа рабочих потоков или процессов.
• Коэффициент ускорения и эффективность: S = T₁/Tₙ, E = S/n.
• Демонстрацию состояния гонки и её устранения блокировкой.
• Объяснение, почему на вашей задаче выигрывает выбранный подход — со ссылкой на замеры.

Закон Амдала — предел ускорения:

  S = 1 / (α + (1 − α)/n)

α — доля последовательной части программы, n — число процессоров

При α = 0,1 (10 % кода не распараллеливается):
  n = 4   → S = 3,08
  n = 16  → S = 6,40
  n = ∞   → S = 10

Вывод, который стоит написать в заключении: даже
бесконечное число ядер не даст больше десятикратного
ускорения, если десятая часть кода последовательна.