Модуль asyncio в Python

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

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

Используяasyncio, можно обрабатывать несколько задач одновременно в одном потоке, не используя несколько потоков или процессов.

Зачем нам нужен asyncio?

В традиционном синхронном программировании, когда задаче нужно дождаться завершения операции ввода-вывода (например, сетевого запроса), программа блокируется до завершения операции. Это приводит к низкой эффективности, особенно при обработке большого количества операций ввода-вывода.

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

Представьте, что вы управляете рестораном:

  • Синхронный режим (обычные функции):У вас только один повар. Гость A заказывает стейк, и повар начинает жарить стейк (для этого нужно подождать 5 минут). В течение этих 5 минут жарки стейка повар полностью занят и не может делать ничего другого. Даже если гость B хочет всего лишь стакан воды, ему приходится ждать.
  • Асинхронный режим (asyncio):У вас несколько поваров (на самом деле всё ещё один, но очень умный). После того как повар начинает жарить стейк гостя A, он понимает, что нужно подождать. Он сразу помечает стейк как ожидающий, затем поворачивается, чтобы налить воды гостю B. Налив воды, он возвращается проверить, почти ли готов стейк. Если он ещё не готов, он может пойти обработать заказ гостя C. Таким образом, во время ожидания ввода-вывода (например, жарки стейка, сетевых запросов, чтения и записи файлов) повар (CPU) всегда работает эффективно.

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

Её ядро — это цикл событий, корутины и задачи.

Основные концепции asyncio

1. Корутина

Корутина — этоasyncioодна из основных концепций. Это особая функция, которая может приостанавливаться во время выполнения и возобновляться позже. Корутины определяются с помощьюasync defключевого слова и, черезawaitключевое слово, используемое для ожидания завершения асинхронной операции.

Пример

import asyncio

async def say_hello():
    print("Hello")
    await asyncio.sleep(1)
    print("World")

2. Цикл событий

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

Пример

async def main():
    await say_hello()

asyncio.run(main())

3. Задача

Задача — это обёртка вокруг корутины, представляющая корутину, которая выполняется или будет выполнена. Вы можете использоватьasyncio.create_task()функцию для создания задачи и добавления её в цикл событий.

Пример

async def main():
    task = asyncio.create_task(say_hello())
    await task

4. Future

Future— это объект, представляющий результат асинхронной операции. Обычно он используется в низкоуровневых API для представления операции, которая ещё не завершена. Вы можете использоватьawaitключевое слово для ожиданияFutureчтобы завершить.

Пример

async def main():
    future = asyncio.Future()
    await future

Базовое использование и примеры кода

Давайте разберём вышеописанные концепции на классическом примере одновременного доступа к нескольким URL.

Предположим, нам нужно получить содержимое трёх разных URL. При синхронном подходе задачи выполняются последовательно, и общее время равно сумме времени трёх запросов. С помощьюasyncio, мы можем отправить эти три запроса одновременно, и общее время будет близко ко времени самого медленного запроса.

Синхронная версия (для сравнения)

Пример

import time
import requests

def fetch_url(url):
    """Имитация длительного сетевого запроса (синхронная версия)"""
    print(f"Начало загрузки: {url}")
    time.sleep(2)  # Имитация 2 секунд сетевой задержки
    print(f"Загрузка завершена: {url}")
    return f"Данные из {url}"

def main_sync():
    urls = ['https://example.com/1', 'https://example.com/2', 'https://example.com/3']
    results = []
    start = time.time()
   
    for url in urls:
        result = fetch_url(url)  # Необходимо дождаться завершения предыдущего, прежде чем запускать следующий
        results.append(result)
   
    end = time.time()
    print(f"Общее время синхронной версии: {end - start:.2f} секунд")
    print(f"Результаты: {results}")

if __name__ == "__main__":
    main_sync()

Ожидаемый вывод：

开始获取: https://example.com/1
完成获取: https://example.com/1
开始获取: https://example.com/2
完成获取: https://example.com/2
开始获取: https://example.com/3
完成获取: https://example.com/3
同步版本总耗时: 6.00 秒
结果: [‘来自 https://example.com/1 的数据‘, ‘来自 https://example.com/2 的数据‘, ‘来自 https://example.com/2 的数据‘]

Это занимает около 6 секунд в общей сложности.

Асинхронная версия (с использованием asyncio)

Нам нужно использоватьaiohttpбиблиотеку для заменыrequestsдля выполнения асинхронных HTTP-запросов. Сначала установите её:pip install aiohttp。

Пример

import asyncio
import aiohttp
import time

async def fetch_url_async(session, url):
    """Имитировать длительный сетевой запрос (асинхронная версия)"""
    print(f"Запуск асинхронной загрузки: {url}")
    # Примечание: Здесь мы используем асинхронный метод get из aiohttp и ожидаем с помощью await
    async with session.get(url) as response:
        # Имитируйте, что обработка ответа тоже занимает время
        await asyncio.sleep(2)  # Используйте asyncio.sleep для имитации ожидания ввода-вывода; это не блокирует поток
        text = await response.text()
        print(f"Асинхронная загрузка завершена: {url}")
        return f"Данные из {url} (длина: {len(text)})"

async def main_async():
    urls = ['https://httpbin.org/get', 'https://httpbin.org/delay/1', 'https://httpbin.org/headers']
   
    async with aiohttp.ClientSession() as session:  # Создайте асинхронную HTTP-сессию
        # Создайте задачу (Task) для каждого URL
        tasks = []
        for url in urls:
            # create_task добавляет корутину в цикл событий и сразу начинает планирование
            task = asyncio.create_task(fetch_url_async(session, url))
            tasks.append(task)
       
        print("Все задачи созданы, запускаем параллельное выполнение...")
       
        # Используйте asyncio.gather, чтобы запустить все задачи параллельно и дождаться их завершения
        # gather возвращает список результатов, порядок которых соответствует порядку переданных задач
        results = await asyncio.gather(*tasks)
       
        return results

if __name__ == "__main__":
    start = time.time()
    # asyncio.run() — удобный способ запустить цикл событий и выполнить корутину верхнего уровня
    final_results = asyncio.run(main_async())
    end = time.time()
   
    print(f"\n"Общее время асинхронной версии: {end - start:.2f} секунд")
    for res in final_results:
        print(res)

Ожидаемый вывод：

所有任务已创建，开始并发执行...
开始异步获取: https://httpbin.org/get
开始异步获取: https://httpbin.org/delay/1
开始异步获取: https://httpbin.org/headers
（大约 2 秒后，所有请求几乎同时完成）
完成异步获取: https://httpbin.org/headers
完成异步获取: https://httpbin.org/get
完成异步获取: https://httpbin.org/delay/1

异步版本总耗时: 2.10 秒  # 注意！总耗时远小于 6 秒
来自 https://httpbin.org/get 的数据 (长度: 274)
来自 https://httpbin.org/delay/1 的数据 (长度: 392)
来自 https://httpbin.org/headers 的数据 (长度: 177)

Анализ кода：

  • async def: определяет функцию-корутинуfetch_url_asyncиmain_async。
  • await: вfetch_url_async, мыawait session.get()иawait response.text(), это сообщает циклу событий: «Этот сетевой запрос требует времени; сначала выполните другие готовые задачи».
  • asyncio.create_task(): оборачиваетfetch_url_asyncкорутину вTask, позволяя ей быть запланированной циклом событий и обеспечивать конкурентность.
  • asyncio.gather(*tasks): очень практичная функция, которая конкурентно запускает все переданные корутины/задачи, ждёт завершения всех их и в итоге собирает все результаты.
  • asyncio.run(main_async()): рекомендуемый подход в Python 3.7+, отвечает за создание цикла событий, запуск корутины и закрытие цикла.

Ключевые функции и описания параметров

В следующей таблице перечисленыasyncioнесколько наиболее часто используемых высокоуровневых функций:

Функция Основное назначение Описание общих параметров
asyncio.run(coro, *, debug=False) Запускает корутину верхнего уровня и управляет жизненным циклом цикла событий. Это главная точка входа в программу. coro: объект корутины для запуска.
debug: установите вTrueчтобы включить режим отладки цикла событий.
asyncio.create_task(coro, *, name=None) Оборачивает корутину в объектTaskи ставит его в очередь в цикле событий для планирования. Это основной способ достижения конкурентности. coro: объект корутины для обёртывания.
name: (Python 3.8+) укажите имя для задачи для упрощения отладки.
asyncio.gather(*aws, return_exceptions=False) Одновременно запускаетнесколько асинхронных задач (awsможет принимать корутины, задачи и т. д.), ожидает завершения всех и возвращает список результатов. *aws: переменные аргументы, передайте несколько асинхронных объектов.
return_exceptions: по умолчанию равноFalse, любое исключение, вызванное задачей, будет немедленно переданоgatherвызывающему. Установите вTrueесли установлено, исключения будут возвращены как часть результатов.
asyncio.sleep(delay, result=None) Асинхронноприостанавливает выполнение на указанное количество секунд. Это ключевое отличие отtime.sleep(блокирующая). delay: Количество секунд для ожидания.
result: Значение, возвращаемое после завершения ожидания.
asyncio.wait(aws, *, timeout=None, return_when=ALL_COMPLETED) Запускает задачи одновременно и ожидает выполнения указанного условия. Возвращает два множества(done, pending), которые представляют собой завершённые и незавершённые задачи соответственно. aws: Коллекция асинхронных объектов.
timeout: Длительность тайм-аута (секунды).
return_when: Определяет, когда возвращать; варианты:FIRST_COMPLETED(первая завершённая),FIRST_EXCEPTION(первое исключение),ALL_COMPLETED(все завершены, по умолчанию).
asyncio.to_thread(func, /, *args, **kwargs) (Python 3.9+) Запускает обычную, потенциально блокирующую синхронную функцию в отдельном потоке и возвращаетawaitawaitable-корутина. Используется для обработки интенсивных вычислений на ЦП или блокирующего ввода-вывода. func: Синхронная функция для запуска в потоке.
*args, **kwargs: Аргументы, переданные функции.

Наглядное представление: схема асинхронного планирования задач

Пояснение к диаграмме: Эта блок-схема показывает, как цикл событий работает подобно диспетчеру. Он поддерживает очередь задач. Когда выполнение задачи доходит доawait(например, ожидания ответа по сети), она приостанавливается. Цикл событий немедленно находит в очереди следующую готовую к выполнению (ready) задачу и выполняет её. Когда операция ввода-вывода приостановленной задачи завершается, цикл событий получает уведомление, меняет статус задачи обратно на ready и продолжает её выполнение в некоторый будущий момент. Таким образом, во время ожидания ввода-вывода процессор полностью используется для выполнения других задач, что обеспечивает конкурентность в рамках одного потока.

Базовое использование asyncio

1. Запуск корутины

Чтобы запустить корутину, можно использоватьasyncio.run()функцию. Она создаёт цикл событий и запускает указанную корутину.

Пример

import asyncio

async def main():
    print("Start")
    await asyncio.sleep(1)
    print("End")

asyncio.run(main())

2. Одновременное выполнение нескольких задач

Вы можете использоватьasyncio.gather()функция для одновременного запуска нескольких корутин и ожидания завершения всех из них.

Пример

import asyncio

async def task1():
    print("Задача 1 запущена")
    await asyncio.sleep(1)
    print("Задача 1 завершена")

async def task2():
    print("Задача 2 запущена")
    await asyncio.sleep(2)
    print("Задача 2 завершена")

async def main():
    await asyncio.gather(task1(), task2())

asyncio.run(main())

3. Управление тайм-аутом

Вы можете использоватьasyncio.wait_for()функция для установки тайм-аута для корутины. Если корутина не завершится в течение указанного времени,asyncio.TimeoutErrorбудет вызвано исключение.

Пример

import asyncio

async def long_task():
    await asyncio.sleep(10)
    print("Задача завершена")

async def main():
    try:
        await asyncio.wait_for(long_task(), timeout=5)
    except asyncio.TimeoutError:
        print("Время ожидания задачи истекло")

asyncio.run(main())

Сценарии применения asyncio

asyncioОсобенно подходит для следующих сценариев:

  1. Сетевые запросы: например, HTTP-запросы, WebSocket-соединение и т. д.
  2. Файловый ввод-вывод: например, асинхронное чтение и запись файлов.
  3. Операции с базой данных: например, асинхронный доступ к базе данных.
  4. Обработка данных в реальном времени: например, обработка очереди сообщений в реальном времени.

Распространённые классы, методы и функции

1. Основные функции

Метод/функция Описание Пример
asyncio.run(coro) Запустить асинхронную главную функцию (Python 3.7+) asyncio.run(main())
asyncio.create_task(coro) Создать задачу и добавить её в цикл событий task = asyncio.create_task(fetch_data())
asyncio.gather(*coros) Запустить несколько корутин одновременно await asyncio.gather(task1, task2)
asyncio.sleep(delay) Асинхронное ожидание (неблокирующее) await asyncio.sleep(1)
asyncio.wait(coros) Управлять тем, как завершаются задачи done, pending = await asyncio.wait([task1, task2])

2. Цикл событий

Метод Описание Пример
loop.run_until_complete(future) Выполнять до завершения задачи loop.run_until_complete(main())
loop.run_forever() Запустить цикл событий бесконечно loop.run_forever()
loop.stop() Остановить цикл событий loop.stop()
loop.close() Закрыть цикл событий loop.close()
loop.call_soon(callback) Запланировать функцию обратного вызова для немедленного выполнения loop.call_soon(print, "Hello")
loop.call_later(delay, callback) Выполнить обратный вызов с задержкой loop.call_later(5, callback)

3. Корутины и задачи

Метод/декоратор Описание Пример
@asyncio.coroutine Декоратор корутины (устаревший, Python 3.4-3.7) @asyncio.coroutine
def old_coro():
async def Определить корутину (Python 3.5+) async def fetch():
task.cancel() Отменить задачу task.cancel()
task.done() Проверить, завершена ли задача if task.done():
task.result() Получить результат задачи (требуется, чтобы задача была завершена) data = task.result()

4. Примитивы синхронизации (похожие наthreading）

Класс Описание Пример
asyncio.Lock() Асинхронная мьютекс-блокировка lock = asyncio.Lock()
async with lock:
asyncio.Event() Уведомление о событии event = asyncio.Event()
await event.wait()
asyncio.Queue() Асинхронная очередь queue = asyncio.Queue()
await queue.put(item)
asyncio.Semaphore() Семафор sem = asyncio.Semaphore(5)
async with sem:

5. Сеть и подпроцессы

Метод/Класс Описание Пример
asyncio.open_connection() Установить TCP-соединение reader, writer = await asyncio.open_connection('host', 80)
asyncio.start_server() Создать TCP-сервер server = await asyncio.start_server(handle, '0.0.0.0', 8888)
asyncio.create_subprocess_exec() Создать подпроцесс proc = await asyncio.create_subprocess_exec('ls')

6. Вспомогательные инструменты

Метод Описание Пример
asyncio.current_task() Получить текущую задачу task = asyncio.current_task()
asyncio.all_tasks() Получить все задачи tasks = asyncio.all_tasks()
asyncio.shield(coro) Защитить задачу от отмены await asyncio.shield(critical_task)
asyncio.wait_for(coro, timeout) Ожидание с тайм-аутом try: await asyncio.wait_for(task, 5)

Пример

1. Базовый пример корутины

Пример

import asyncio

async def hello():
    print("Hello")
    await asyncio.sleep(1)
    print("World")

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

2. Одновременное выполнение задач

Пример

async def fetch(url):
    print(f"Fetching {url}")
    await asyncio.sleep(2)
    return f"Данные из {url}"

async def main():
    results = await asyncio.gather(
        fetch("url1.com"),
        fetch("url2.com")
    )
    print(results)

asyncio.run(main())

3. Использование асинхронной очереди

Пример

async def producer(queue):
    for i in range(5):
        await queue.put(i)
        await asyncio.sleep(0.1)

async def consumer(queue):
    while True:
        item = await queue.get()
        print(f"Consumed {item}")
        queue.task_done()

async def main():
    queue = asyncio.Queue()
    await asyncio.gather(
        producer(queue),
        consumer(queue)
    )

Примечания

  1. Версия Python: Некоторые функции требуют Python 3.7+ (например,asyncio.run()）。

  2. Блокирующие операции: Избегайте использования синхронного блокирующего кода в корутинах (например,time.sleep()）。

  3. Отладка: УстановитеPYTHONASYNCIODEBUG=1переменную среды, чтобы включить режим отладки.

  4. Отмена задач: Отменённая задача вызоветCancelledError, что необходимо правильно обработать.

Другие расширения