Python: полный мануал с примерами и справочником функций

Глава 22 · продвинутый

Asyncio, потоки, процессы и GIL

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

Конкурентность не всегда означает параллельный расчёт

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

Async def создаёт корутину, await выполняет ожидание

Вызов async-функции возвращает объект корутины. Для её выполнения нужен await либо создание задачи в работающем цикле. Несколько последовательных await не превращаются автоматически в конкурентный запуск. В примере задачи создаются вместе внутри TaskGroup, доступного с Python 3.11. Список задач сохраняет выбранный порядок результатов; порядок фактического завершения может быть другим. Небольшое asyncio.sleep лишь моделирует освобождающее цикл ожидание, а не ускоряет сам расчёт квадратов.

import asyncio

async def calculate(value):
    await asyncio.sleep(0)
    return value * value

async def main():
    async with asyncio.TaskGroup() as group:
        tasks = [group.create_task(calculate(value)) for value in [3, 1, 2]]
    print("Результаты:", [task.result() for task in tasks])

asyncio.run(main())

Результаты: [9, 1, 4]

Обычный time.sleep внутри корутины блокирует её поток вместе с циклом событий. То же происходит при длительном синхронном чтении или вычислении без подходящего освобождения управления. Для блокирующего I/O можно использовать asyncio.to_thread, помня о природе библиотеки и отмене. Асинхронный клиент должен поддерживать эту модель сам. async with и async for выражают асинхронное управление ресурсом и итерацию; добавление ключевого слова к обычному объекту не делает его автоматически асинхронным.

Отмена должна пройти через cleanup

TaskGroup связывает время жизни созданных задач: выход ожидает их, а существенная ошибка приводит к отмене соседних задач и обработке группы исключений. Отмена доставляется в точках ожидания как CancelledError; освобождайте ресурс в finally и обычно пропускайте отмену дальше. Если поймать её и продолжить бесконечную работу, таймаут и группа перестанут вести себя ожидаемо. Таймаут ограничивает ожидание по модели asyncio, но не способен насильно остановить произвольный блокирующий код, который не возвращает управление.

import asyncio

async def worker(started, cleaned):
    started.set()
    try:
        await asyncio.Event().wait()
    finally:
        cleaned.set()

async def main():
    started, cleaned = asyncio.Event(), asyncio.Event()
    task = asyncio.create_task(worker(started, cleaned))
    await started.wait()
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("Отмена получена")
    print("Ресурс освобождён:", cleaned.is_set())

asyncio.run(main())

Отмена получена
Ресурс освобождён: True

Потоки: передавайте результаты через очередь

Потоки работают внутри одного процесса и разделяют его память. Это удобно для ожидания, но создаёт гонки: два участника могут изменить одно состояние в неподходящий момент. queue.Queue обеспечивает безопасный канал передачи между потоками. В примере каждый работник отдаёт готовую пару, главный поток ждёт завершения через join и сортирует результат. Размер очереди и число работников нужно ограничивать для большого потока задач; иначе скорость поступления работы может оказаться выше скорости обработки.

from queue import Queue
from threading import Thread, Lock

results = Queue()
lock = Lock()
completed = 0

def work(value):
    global completed
    results.put((value, value * value))
    with lock:
        completed += 1

threads = [Thread(target=work, args=(value,)) for value in [3, 1, 2]]
for thread in threads:
    thread.start()
for thread in threads:
    thread.join()
collected = [results.get() for _ in threads]
print("Результаты:", sorted(collected))
print("Завершено:", completed)

Результаты: [(1, 1), (2, 4), (3, 9)]
Завершено: 3

Lock защищает выбранную критическую секцию, а не весь смысл программы сам по себе. Договоритесь, какие данные он охраняет и в каком порядке берутся несколько блокировок. Не удерживайте блокировку во время долгого сетевого запроса, если это не требуется контрактом: остальные работники будут ждать. GIL не заменяет такую защиту составного действия. Ошибка внутри обычного потока также не обязана автоматически подняться в главном: передавайте её через явный канал либо используйте Future с вызовом result.

Для процессов нужен импортируемый код

Процессы имеют отдельную память и передают сериализуемые данные. Для функций пула используйте определения уровня модуля, а не локальную функцию или lambda. Защитите запуск через __main__. В примере явно выбран spawn и функция находится в соседнем модуле; это показывает переносимое устройство, а не предполагает старое наследование состояния через fork. На Windows и macOS обычно используется spawn; с Python 3.14 стандартный выбор POSIX изменился на forkserver там, где он поддерживается.

from concurrent.futures import ProcessPoolExecutor
from multiprocessing import get_context
from manual_workers import square

def main():
    with ProcessPoolExecutor(max_workers=2, mp_context=get_context("spawn")) as pool:
        values = list(pool.map(square, [1, 2, 3, 4]))
    print("Квадраты:", values)

if __name__ == "__main__":
    main()

Квадраты: [1, 4, 9, 16]

Не передавайте в процесс открытое соединение, заблокированный ресурс или огромный объект без оценки стоимости. Разделите работу на разумные пачки и выбирайте число работников по измерению, а не по максимальному доступному числу ядер. Executor.map отдаёт результаты по порядку входа; as_completed позволяет обработать завершившиеся раньше. Ошибка проявится при получении соответствующего результата. Запрашивая отмену Future, не рассчитывайте, что уже начавшаяся операция немедленно прервётся: это зависит от её состояния и модели.

Обычная и free-threaded сборки различаются

В обычной сборке CPython GIL ограничивает одновременное исполнение Python-кода потоками одного интерпретатора, но потоки по-прежнему полезны для I/O; некоторые расширения освобождают GIL при вычислении. Free-threaded сборка появилась в Python 3.13 и получила официальный статус поддержки в 3.14. Наличие такой сборки и фактически выключенный GIL — разные факты: несовместимое расширение может включить его обратно. Проверки sysconfig.get_config_var("Py_GIL_DISABLED") и sys._is_gil_enabled() отвечают на разные вопросы.

Изолированные интерпретаторы и контекст задач

InterpreterPoolExecutor и concurrent.interpreters появились в Python 3.14. Они предлагают отдельные интерпретаторы с изолированным состоянием; это не обычные потоки с общей копией всех изменяемых объектов. Учитывайте передачу данных, импорты и совместимость расширений. contextvars помогает хранить контекст логической задачи, например учебный идентификатор операции, не смешивая его между корутинами. Это средство передачи контекста, а не блокировка общей базы. Выбирайте новый механизм по реальной пользе и измерению, сохраняя более простой путь там, где он достаточен.

Практика: результат без гонки вывода

Запустите три функции через ThreadPoolExecutor, верните значения и напечатайте итог только из главного потока. Не проверяйте порядок промежуточных print работников: планировщик не обещает такого порядка. Затем замените одну функцию на вызывающую ValueError и убедитесь, что result или обход map передаёт ошибку главному коду. Это практичная граница между параллельным выполнением и последовательным оформлением отчёта.

from concurrent.futures import ThreadPoolExecutor

def prepare(value):
    return value.upper()

with ThreadPoolExecutor(max_workers=2) as pool:
    results = list(pool.map(prepare, ["one", "two", "three"]))
print(", ".join(results))

ONE, TWO, THREE