Содержание страницы

asyncio и свободно-поточный Python¶asyncio and free-threaded Python

asyncio использует цикл событий в качестве планировщика, чтобы обеспечить высокоэффективную параллельность за счёт переключения между задачами для неблокирующих операций ввода-вывода. Это даёт лучшую производительность для сценариев, связанных с вводом-выводом. Он также позволяет выгружать процессороёмкие задачи в пул потоков или процессов, но это по-прежнему ограничено глобальной блокировкой интерпретатора в CPython.

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

Начиная с Python 3.14, asyncio имеет первоклассную поддержку свободно-поточного Python, и реализация asyncio безопасна для использования в многопоточной среде.

Один цикл событий на одном ядре может обрабатывать множество соединений параллельно, но код Python, выполняемый для обработки каждого из них, всё равно выполняется последовательно. Как только запросы требуют нетривиального объёма вычислений на один запрос, обработка становится узким местом, и одно ядро больше не справляется. Объединение asyncio с потоками наиболее полезно здесь: запуская цикл событий на поток, обработка разных запросов может выполняться параллельно на нескольких ядрах CPU. Это также полезно, когда нужно выполнить блокирующий или процессороёмкий код из приложения asyncio.

См. также

Масштабирование asyncio в свободно-поточном Python, статья в блоге Кумара Адитьи, в которой объясняются внутренние изменения, делающие asyncio безопасным и эффективным в свободно-поточном Python, вместе с бенчмарками полученных улучшений.

Вопросы потокобезопасности¶Thread safety considerations

Хотя asyncio разработан как потокобезопасный в среде свободно-поточного Python, всё же есть некоторые моменты, которые следует учитывать при использовании asyncio с потоками:

Цикл событий: Каждый поток должен иметь свой собственный цикл событий, который не должен использоваться совместно с другими потоками. Это гарантирует, что цикл событий может управлять своими задачами и колбэками без вмешательства со стороны других потоков.
Управление задачами: Задачи и future, созданные в одном потоке, не должны ожидаться или изменяться из другого потока.
Потокобезопасные API: При взаимодействии с asyncio из нескольких потоков важно использовать потокобезопасные API, предоставляемые asyncio, например asyncio.run_coroutine_threadsafe() для отправки корутин в цикл событий из другого потока. Если нужно вызвать колбэк из другого потока, можно использовать loop.call_soon_threadsafe() для безопасного планирования.
Синхронизация: Примитивы синхронизации, предоставляемые asyncio (например, asyncio.Lock и asyncio.Event), не предназначены для использования между потоками. Если нужно синхронизировать потоки, следует использовать примитивы синхронизации из модуля threading.

Использование asyncio с потоками¶Using asyncio with threads

asyncio поддерживает запуск одного цикла событий на поток, что позволяет задействовать несколько ядер CPU в среде свободно-поточного Python. Каждый поток может запускать свой собственный цикл событий, и задачи могут планироваться в этих циклах независимо.

Вот пример использования asyncio с потоками:

import asyncio
import threading

async def worker(name: str) -> None:
    print(f"Worker {name} starting")
    await asyncio.sleep(1)
    print(f"Worker {name} done")

def run_loop(name: str) -> None:
    asyncio.run(worker(name))

threads = [
    threading.Thread(target=run_loop, args=(f"T{i}",))
    for i in range(4)
]
for t in threads:
    t.start()
for t in threads:
    t.join()

В этом примере каждый поток создаёт свой собственный цикл событий с помощью asyncio.run() и запускает на нём корутину. Потоки выполняются одновременно, и в сборке со свободными потоками они могут выполняться на разных ядрах CPU параллельно.

Производитель/потребитель между потоками¶Producer/consumer across threads

Когда обычному (не asyncio) потоку нужно передать работу циклу событий asyncio, работающему в другом потоке, следует использовать потокобезопасный примитив, например queue.Queue, а не asyncio.Queue, который безопасен только внутри одного цикла событий:

import asyncio
import queue
import threading

def producer(q: queue.Queue[int]) -> None:
    for i in range(5):
        print(f"Producing {i}")
        q.put(i)
    q.shutdown()

async def consumer(q: queue.Queue[int]) -> None:
    while True:
        try:
            item = q.get_nowait()
        except queue.Empty:
            await asyncio.sleep(0.1)
            continue
        except queue.ShutDown:
            break
        print(f"Consumed {item}")
        await asyncio.sleep(item)

q: queue.Queue[int] = queue.Queue()
consumer_thread = threading.Thread(
    target=lambda: asyncio.run(consumer(q))
)
consumer_thread.start()
producer(q)
consumer_thread.join()

Производитель выполняется в главном потоке, а потребитель – внутри цикла событий в своём собственном потоке, но они безопасно взаимодействуют через queue.Queue. Когда очередь пуста, потребитель ненадолго засыпает и повторяет попытку. Когда производитель завершает работу, он вызывает shutdown(), что заставляет последующие вызовы get_nowait() порождать queue.ShutDown, чтобы потребитель мог корректно завершиться.