Руководство по асинхронному программированию в Python

Когда вы погружаетесь в мир Python, то замечаете, что asyncio - это настоящая находка для решения задач по веб разработке и большинства сетевых задач. Этот набор инструментов является ответом Python на необходимость написания чистого, эффективного и масштабируемого кода для конкурентных (параллельных) операций ввода-вывода. Поначалу это может показаться немного пугающим, с его циклами событий, сопрограммами (корутинами) и фьючерсами. Но как только вы освоитесь с этим, вы будете удивляться, как раньше жили без этого. Итак, давайте разберем это шаг за шагом, с примерами и взглянем на то, как все выглядит по другую сторону асинхронного барьера.

Принцип работы asyncio

Прежде чем перейти к примерам, важно понять основные концепции asyncio:

1. Цикл обработки событий (Event loop): центральное исполнительное устройство, предоставляемое asyncio. Оно управляет и распределяет выполнение различных задач. Оно отвечает за обработку событий и планирование асинхронных процедур (routine).
2. Сопрограммы (coroutines): Асинхронные функции, объявленные с помощью async def. Эти функции могут быть приостановлены и возобновлены в моменты ожидания, что позволяет выполнять операции ввода-вывода в фоновом режиме.
3. Будущие объекты или фьючерсы (futures): объекты, представляющие результат работы, которая еще не была завершена. Они возвращаются из задач, запланированных циклом обработки событий.
4. Задачи (tasks): запланированные сопрограммы, которые переносятся в будущий объект с помощью цикла обработки событий, что позволяет их выполнять.

Начало работы с Asyncio

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

Ключевое слово await

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

Что представляет из себя await

1. Контекст: await может использоваться только внутри асинхронных функций. Попытка использовать его вне такого контекста приводит к синтаксической ошибке.
2. Назначение: основная цель - вернуть управление циклу обработки событий, приостановив выполнение включающей его сопрограммы до тех пор, пока не будет разрешен ожидаемый объект. Это неблокирующее поведение делает асинхронное программирование эффективным, особенно для задач, связанных с вводом/выводом.
3. Ожидаемые объекты: объекты, которые могут использоваться с await, должны быть доступны для ожидания. Наиболее распространенные ожидаемые объекты - это сопрограммы, объявленные с помощью async def, но и другие включают задачи asyncio, фьючерсы или любой объект с методом __await__().

Примеры

Асинхронный Привет мир!

Представьте, что вам нужно напечатать "Привет, мир!" после 2-секундной паузы. Синхронный подход прост:

import time

def say_hello():
    time.sleep(2)
    print("Привет мир!")

say_hello()

Он выполняет свою работу, но во время ожидания этих 2 секунд все останавливается.

Теперь давайте переключимся на asyncio, используя асинхронный способ:

import asyncio

async def say_hello_async():
    await asyncio.sleep(2)
    print("Привет мир!")

asyncio.run(say_hello_async())

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

import asyncio

async def say_hello_async():
    await asyncio.sleep(2)  # Симуляция ожидания в 2 сек.
    print("Привет мир!")

async def do_something_else():
    print("Выполняем другую задачу...")
    await asyncio.sleep(1)  # Симуляция работы другой задачи в 1 сек.
    print("Другая задача выполнена!")

async def main():
    # Запланируем одновременное выполнение
    # обеих задач
    await asyncio.gather(
        say_hello_async(),
        do_something_else(),
    )

asyncio.run(main())

В коде функция main() использует asyncio.gather() для одновременного запуска say_hello_async() и do_something_else(). Это означает, что пока программа ожидает завершения 2-секундного ожидания функции say_hello_async(), она запускает и потенциально завершает функцию do_something_else(), эффективно выполняя другую задачу в течение времени ожидания.

Загрузка веб-страниц (параллельные задачи ввода-вывода)

Загрузка веб-страниц - классический пример, демонстрирующий возможности асинхронного программирования. Давайте сравним загрузку URL-адресов синхронно и асинхронно.

Синхронные HTTP-запросы в основном выполняются библиотекой requests, загрузка двух веб-страниц подряд выглядит примерно так:

import requests
import time

start_time = time.time()

def fetch(url):
    return requests.get(url).text

page1 = fetch('http://example.com')
page2 = fetch('http://example.org')

print(f"Завершено за {time.time() - start_time} секунд")

# Вывод: Завершено за 0.6854028701782227 секунд

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

Давайте повысим эффективность с помощью aiohttp и asynci, которые можно использовать для асинхронных HTTP-запросов:

import aiohttp
import asyncio
import time

async def fetch_async(url, session):
    async with session.get(url) as response:
        return await response.text()

async def main():
    async with aiohttp.ClientSession() as session:
        page1 = asyncio.create_task(fetch_async('http://example.com', session))
        page2 = asyncio.create_task(fetch_async('http://example.org', session))
        await asyncio.gather(page1, page2)

start_time = time.time()
asyncio.run(main())
print(f"Завершено за {time.time() - start_time} секунд")

# Вывод: Завершено за 0.3990549087524414 секунд

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

Чтение файлов (параллельные задачи ввода-вывода)

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

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

# Синхронное чтение нескольких файлов
def read_file_sync(filepath):
    with open(filepath, 'r') as file:
        return file.read()

def read_all_sync(filepaths):
    return [read_file_sync(filepath) for filepath in filepaths]

filepaths = ['file1.txt', 'file2.txt']
data = read_all_sync(filepaths)
print(data)

Для асинхронной версии мы будем использовать aiofiles, библиотеку, которая обеспечивает поддержку асинхронных файловых операций. Если вы еще не установили aiofiles, вы можете сделать это с помощью pip:

pip install aiofiles

С помощью aiofiles мы можем выполнять операции ввода-вывода файлов, не блокируя цикл обработки событий, что позволяет нам считывать несколько файлов одновременно.

import asyncio
import aiofiles

# Асинхронное чтение одного файла
async def read_file_async(filepath):
    async with aiofiles.open(filepath, 'r') as file:
        return await file.read()

async def read_all_async(filepaths):
    tasks = [read_file_async(filepath) for filepath in filepaths]
    return await asyncio.gather(*tasks)

# Запуск асинхронной функции
async def main():
    filepaths = ['file1.txt', 'file2.txt']
    data = await read_all_async(filepaths)
    print(data)

asyncio.run(main())

Асинхронная версия, использующая aiofiles и asyncio.gather, позволяет выполнять одновременное чтение нескольких файлов. Такой подход значительно сокращает общее время выполнения по сравнению с синхронной версией, в которой каждый файл считывается один за другим. Выполняя операции ввода-вывода одновременно, мы можем повысить эффективность программ, которым необходимо выполнять несколько файловых операций.

Сочетание асинхронности и синхронизации: гибридный подход

Иногда вы не можете избежать использования синхронных функций, но все же хотите насладиться асинхронной работой. Вот как вы можете их сочетать:

import asyncio
import time

def sync_task():
    print("Запуск медленной синхронной задачи...")
    time.sleep(5)  # Имитация долгой работы
    print("Завершено выполнение задачи.")

async def async_wrapper():
    loop = asyncio.get_running_loop()
    await loop.run_in_executor(None, sync_task)

async def main():
    await asyncio.gather(
        async_wrapper(),
        # Здесь будут другие асинхронные задачи
    )

asyncio.run(main())

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

Разбор кода:

1. Асинхронная обертка (функция async_wrapper):

1. Эта асинхронная функция демонстрирует, как запускать синхронную sync_task таким образом, чтобы не блокировать цикл обработки событий. Это достигается с помощью loop.run_in_executor(None, sync_task).
2. loop.run_in_executor(None, sync_task) запускает sync_task в отдельном потоке или процессе, в зависимости от используемого исполнителя (executor). Исполнитель по умолчанию (None, указанный в качестве первого аргумента) запускает задачи в пуле потоков.
3. await используется для ожидания завершения sync_task без блокировки цикла обработки событий, позволяя тем временем выполнять другие асинхронные операции.

2. Асинхронное выполнение (функция main()):

1. Асинхронная функция main() демонстрирует, как выполнять синхронные и асинхронные задачи одновременно без блокировки.
2. asyncio.gather используется для планирования параллельного выполнения async_wrapper и, возможно, других асинхронных задач. Используя gather, вы гарантируете, что цикл обработки событий может управлять несколькими задачами, выполняя их одновременно, где это возможно.

3. Запуск цикла обработки событий (asyncio.run(main())):

1. В завершение, asyncio.run(main()) вызывается для запуска основной сопрограммы, которая эффективно запускает цикл обработки событий и выполняет задачи, запланированные в main.

Зачем нужен такой подход?

1. Интеграция устаревшего (легаси) кода: В реальных приложениях вы часто сталкиваетесь с устаревшим кодом, который является синхронным по своей природе. Переписывание больших кодовых баз для обеспечения асинхронной совместимости не всегда возможно. Такой подход позволяет легко интегрировать такой код в ваши асинхронные приложения.
2. Работа с блокирующим вводом-выводом: Некоторые операции, особенно те, которые связаны с блокирующим вводом-выводом, не имеют асинхронных эквивалентов, или вы, возможно, работаете со сторонними библиотеками, которые предлагают только синхронные функции. Этот метод позволяет перенести эти операции в поток, освобождая цикл обработки событий для обработки других асинхронных задач.
3. Задачи, связанные с процессором: Хотя задачи, связанные с процессором, обычно лучше решаются с помощью многопроцессорной обработки из-за глобальной блокировки интерпретатора Python (GIL), иногда вы можете запускать их в потоках для простоты или потому, что вычислительные затраты не слишком высоки. Использование run_in_executor позволяет этим задачам сосуществовать с асинхронными задачами, связанными с вводом-выводом.

Объект Future()

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

Механизм работы Фьючерсов

1. Роль: Фьючерсы используются для объединения низкоуровневых асинхронных операций с высокоуровневыми приложениями asyncio. Они предоставляют способ управления состоянием асинхронной операции: ожидание, завершение (с результатом) или сбой (с исключением).
2. Использование: Как правило, вам не нужно самостоятельно создавать фьючерсы при использовании высокоуровневых функций и конструкций asyncio (например, задачи (таски) которые являются подклассом Future). Однако понимание фьючерсов важно для взаимодействия с асинхронными API более низкого уровня или при построении сложных асинхронных систем.

Работа с фьючерсами

Объект Future имеет несколько ключевых методов и свойств:

1. set_result(result): устанавливает результат Future. Это отметит его как выполненный и уведомит все выполняемые сопрограммы.
2. set_exception(exception): устанавливает исключение как результат Future. Это также отметит его как выполненное, но вызовет исключение, когда оно будет выполняться.
3. add_done_callback(callback): Добавляет функцию обратного вызова, которая будет вызываться, когда Future завершится (либо завершится с результатом, либо с исключением).
4. result(): Возвращает результат Future. Если Future не выполняется, то возникает ошибка InvalidStateError. Если Future завершается с исключением, этот метод повторно вызывает исключение.
5. done(): Возвращает значение True, если Future выполнен. Future считается выполненным, если оно имеет результат или исключение.

import asyncio

# Функция для моделирования асинхронной
# операции с использованием Future
async def async_operation(future, data):
    await asyncio.sleep(1)  # Имитация асинхронной работы с задержкой
    
    # Установим результат или исключение
    # на основе входных данных
    if data == "success":
        future.set_result("Операция прошла успешно")
    else:
        future.set_exception(RuntimeError("Операция выполнена с ошибкой"))

# Функция будет вызвана, когда Future будет завершен
def future_callback(future):
    try:
        print("Обратная связь:", future.result())  # Попытка вывести результат
    except Exception as exc:
        print("Обратная связь:", exc)  # Выводит исключение если ошибка

async def main():
    # Создание объекта Future
    future = asyncio.Future()
    
    # Добавим функцию обратной связи для Future
    future.add_done_callback(future_callback)
    
    # Запустим асинхронную опреацию и передадим Future
    await async_operation(future, "успешно")  # Пробуем изменить "успешно" на что-то другое для имитации ошибки
    
    # Проверяем что Future выполнен
    # и выводим его результат
    if future.done():
        try:
            print("Main:", future.result())
        except Exception as exc:
            print("Main:", exc)

# Запускаем сопрограмму Main
asyncio.run(main())

Как это работает

1. async_operation - это асинхронная функция, имитирующая асинхронную задачу, которая принимает будущий объект и некоторые данные в качестве аргументов. Она ожидает 1 секунду, чтобы имитировать асинхронную работу. Основываясь на значении данных, он либо устанавливает результат во Future, используя set_result, либо создает исключение, используя set_exception.
2. future_callback - это функция обратной связи, которая выводит результат Future после его завершения. Она проверяет, была ли операция выполнена успешно или нет, вызывая функцию future.result(), которая либо возвращает результат, либо повторно вызывает исключение, установленное во Future.
3. В основной сопрограмме создается объект Future, и future_callback добавляется в качестве его обратного вызова с помощью add_done_callback. Затем выполняется async_operation с данными Future и аргументами ("успешно" или любое другое значение для имитации сбоя).
4. После завершения async_operation, main проверяет, выполнено ли Future с помощью done(). Затем он пытается напечатать результат напрямую, обрабатывая любые возможные исключения.

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

Заключение

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