Асинхронное программирование на Python: практические приёмы asyncio

Расширенные примеры использования asyncio

Пример 1: Отмена задачи через таймаут

import asyncio

async def long_running():
    try:
        await asyncio.sleep(10)
        return 'Done'
    except asyncio.CancelledError:
        print('Задача отменена')
        raise

async def main():
    task = asyncio.create_task(long_running())
    try:
        await asyncio.wait_for(task, timeout=2)
    except asyncio.TimeoutError:
        print('Таймаут, задача отменяется')
        if not task.done():
            task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            pass

if __name__ == '__main__':
    asyncio.run(main())

Таймаут, задача отменяется
Задача отменена

Здесь wait_for поднимает TimeoutError по истечении лимита. В блоке except явно отменяем задачу и обрабатываем CancelledError внутри самой корутины для корректной очистки.

Пример 2: Защита критической секции от отмены с asyncio.shield

import asyncio

async def critical_operation():
    print('Начало критической операции')
    await asyncio.sleep(2)
    print('Конец критической операции')
    return 'success'

async def main():
    shielded = asyncio.shield(critical_operation())
    try:
        await asyncio.wait_for(shielded, timeout=1)
    except asyncio.TimeoutError:
        print('Истёк таймаут, но операция продолжается')
        # Операция не отменена, можем дождаться её позже
        result = await shielded
        print('Результат защищённой операции:', result)

if __name__ == '__main__':
    asyncio.run(main())

Начало критической операции
Истёк таймаут, но операция продолжается
Конец критической операции
Результат защищённой операции: success

shield оборачивает корутину так, что её нельзя отменить извне. Полезно для завершения обязательных действий, например записи в базу данных.

Пример 3: Асинхронные итераторы и генераторы

import asyncio

class AsyncCountdown:
    def __init__(self, start):
        self.count = start

    def __aiter__(self):
        return self

    async def __anext__(self):
        if self.count <= 0:
            raise StopAsyncIteration
        await asyncio.sleep(0.5)
        self.count -= 1
        return self.count + 1

async def main():
    async for number in AsyncCountdown(3):
        print(number)

if __name__ == '__main__':
    asyncio.run(main())

3
2
1

Асинхронный итератор реализует методы __aiter__ и __anext__. Используется с async for.

Пример 4: Использование asyncio.Queue с приоритетом

import asyncio
import random

class PriorityQueue(asyncio.PriorityQueue):
    pass

async def producer(queue):
    for i in range(5):
        priority = random.randint(1, 5)
        await queue.put((priority, f'item {i}'))
        await asyncio.sleep(0.3)
    await queue.put((0, None))

async def consumer(queue):
    while True:
        priority, item = await queue.get()
        if item is None:
            break
        print(f'Обработан {item} с приоритетом {priority}')

async def main():
    pq = PriorityQueue()
    await asyncio.gather(producer(pq), consumer(pq))

if __name__ == '__main__':
    asyncio.run(main())

Обработан item ... с приоритетом ... (порядок по возрастанию приоритета)

asyncio.PriorityQueue извлекает элементы с наименьшим приоритетом (число). Позволяет реализовать справедливую обработку.

Пример 5: Комбинация asyncio и многопоточности через run_in_executor с ProcessPoolExecutor

import asyncio
from concurrent.futures import ProcessPoolExecutor
import math

def heavy_cpu(x):
    return math.factorial(x)

async def main():
    loop = asyncio.get_running_loop()
    with ProcessPoolExecutor() as pool:
        tasks = [loop.run_in_executor(pool, heavy_cpu, i) for i in range(1000, 1010)]
        results = await asyncio.gather(*tasks)
        print(results)

if __name__ == '__main__':
    asyncio.run(main())

Для CPU-затратных задач используем ProcessPoolExecutor, чтобы обойти GIL. Результат - вычисление факториалов в параллельных процессах.

Пример 6: Асинхронный контекстный менеджер (подключение к базе)

import asyncio

class AsyncDatabase:
    async def connect(self):
        print('Подключение к БД')
        await asyncio.sleep(0.1)
        return self

    async def query(self, sql):
        print(f'Выполнение {sql}')
        await asyncio.sleep(0.1)
        return 'data'

    async def close(self):
        print('Закрытие соединения')

    async def __aenter__(self):
        return await self.connect()

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.close()

async def main():
    async with AsyncDatabase() as db:
        result = await db.query('SELECT 1')
        print(result)

if __name__ == '__main__':
    asyncio.run(main())

Подключение к БД
Выполнение SELECT 1
data
Закрытие соединения

Контекстный менеджер с async with гарантирует корректное открытие и закрытие ресурса.

Пример 7: Обработка множества сокетов с asyncio (эхо-сервер)

import asyncio

async def handle_client(reader, writer):
    addr = writer.get_extra_info('peername')
    print(f'Подключен {addr}')
    while True:
        data = await reader.read(100)
        if not data:
            break
        writer.write(data)
        await writer.drain()
    print(f'Отключен {addr}')
    writer.close()

async def main():
    server = await asyncio.start_server(handle_client, '127.0.0.1', 8888)
    async with server:
        await server.serve_forever()

if __name__ == '__main__':
    asyncio.run(main())

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