Как запустить параллельно две задачи в асинхронном программировании?

Асинхронное программирование в Python

Асинхронное программирование (async/await) позволяет выполнять множество I/O операций одновременно в одном потоке, не блокируя основную программу. Это особенно полезно для сетевых запросов, операций с БД, файлами.

Способ 1: asyncio.gather() (рекомендуется)

gather() запускает несколько корутин одновременно и ждёт, пока все завершатся.

import asyncio

async def task1():
    print("Задача 1 начата")
    await asyncio.sleep(2)
    print("Задача 1 завершена")
    return "Результат 1"

async def task2():
    print("Задача 2 начата")
    await asyncio.sleep(3)
    print("Задача 2 завершена")
    return "Результат 2"

async def main():
    # Запускаем обе задачи параллельно
    results = await asyncio.gather(task1(), task2())
    print(f"Результаты: {results}")
    # Вывод: ["Результат 1", "Результат 2"]
    # Время выполнения: ~3 секунды (а не 5)

asyncio.run(main())

Плюсы:

• Простой и интуитивный синтаксис
• Возвращает результаты в исходном порядке
• Если одна задача падает, можно управлять ошибками

Минусы:

• Ждёт всех задач (даже если одна упала)

Обработка исключений с gather()

async def main():
    try:
        results = await asyncio.gather(
            task1(),
            task2(),
            task3(),
            return_exceptions=True  # Не выбрасывает исключение
        )
        print(results)  # [результат1, Exception(...), результат3]
    except Exception as e:
        print(f"Ошибка: {e}")

Способ 2: asyncio.create_task() (более гибкий)

Создаёт задачу, которая выполняется в фоне.

async def main():
    # Создаём две задачи
    task1_obj = asyncio.create_task(task1())
    task2_obj = asyncio.create_task(task2())
    
    # Ждём обе задачи
    result1 = await task1_obj
    result2 = await task2_obj
    
    print(f"Результаты: {result1}, {result2}")

asyncio.run(main())

Отличие от gather():

• Больше контроля над отдельными задачами
• Можно отменить задачу: task1_obj.cancel()
• Можно проверить статус: task1_obj.done()

async def main():
    task1_obj = asyncio.create_task(task1())
    task2_obj = asyncio.create_task(task2())
    
    # Проверяем, готовы ли задачи
    await asyncio.sleep(1)
    print(f"Task1 завершена? {task1_obj.done()}")  # False
    print(f"Task2 завершена? {task2_obj.done()}")  # False
    
    # Отменяем вторую задачу
    task2_obj.cancel()
    
    result1 = await task1_obj  # Ждём первую
    # await task2_obj  # Выбросит CancelledError

Способ 3: asyncio.wait() (максимум контроля)

Позволяет ждать задач с различными условиями: first_completed, first_exception, all_completed.

async def main():
    task1_obj = asyncio.create_task(task1())
    task2_obj = asyncio.create_task(task2())
    
    # Ждём, когда хотя бы одна завершится
    done, pending = await asyncio.wait(
        [task1_obj, task2_obj],
        return_when=asyncio.FIRST_COMPLETED
    )
    
    print(f"Завершённые: {len(done)}")  # 1
    print(f"В процессе: {len(pending)}")  # 1
    
    # Обработаем результаты завершённых
    for task in done:
        print(f"Результат: {task.result()}")
    
    # Ждём остальные
    done, pending = await asyncio.wait(pending)
    for task in done:
        print(f"Результат: {task.result()}")

asyncio.run(main())

Варианты return_when:

• asyncio.FIRST_COMPLETED — вернуть, когда завершится первая
• asyncio.FIRST_EXCEPTION — вернуть при первой ошибке
• asyncio.ALL_COMPLETED — вернуть, когда все завершены (по умолчанию)

Способ 4: asyncio.TaskGroup() (Python 3.11+)

Современный подход с автоматической очисткой и обработкой ошибок.

import asyncio

async def main():
    async with asyncio.TaskGroup() as tg:
        # Создаём две задачи в группе
        task1 = tg.create_task(task1())
        task2 = tg.create_task(task2())
    
    # После выхода из контекста — все задачи завершены
    print(f"Результат 1: {task1.result()}")
    print(f"Результат 2: {task2.result()}")

asyncio.run(main())

Преимущества:

• Автоматически отменяет оставшиеся задачи при ошибке
• Проще обработка исключений
• Более безопасен

Реальный пример: параллельные HTTP запросы

import aiohttp
import asyncio

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

async def fetch_multiple(urls):
    async with aiohttp.ClientSession() as session:
        # Способ 1: gather()
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        return results
        
        # Или способ 2: TaskGroup (Python 3.11+)
        # async with asyncio.TaskGroup() as tg:
        #     tasks = [tg.create_task(fetch_url(session, url)) for url in urls]
        # return [task.result() for task in tasks]

async def main():
    urls = [
        'https://api.github.com',
        'https://api.example.com',
        'https://httpbin.org/delay/2'
    ]
    
    # Параллельно запросим все URLs
    results = await fetch_multiple(urls)
    
    for i, content in enumerate(results):
        print(f"URL {i}: {len(content)} символов")

asyncio.run(main())

Сравнение подходов

Частые ошибки

❌ Неправильно: без await

# Это НЕ запустит задачи параллельно!
result1 = task1()  # Просто создаёт корутину, не выполняет
result2 = task2()

✅ Правильно

results = await asyncio.gather(task1(), task2())

❌ Неправильно: синхронные операции

async def bad_task():
    time.sleep(2)  # Блокирует весь event loop!
    return "Done"

# Используйте asyncio.sleep() вместо time.sleep()
async def good_task():
    await asyncio.sleep(2)  # Не блокирует
    return "Done"

Производительность

import time
import asyncio

# 3 задачи по 2 секунды

# Синхронно (последовательно): ~6 секунд
start = time.time()
for _ in range(3):
    time.sleep(2)
print(f"Синхронно: {time.time() - start:.2f}s")

# Асинхронно (параллельно): ~2 секунды
async def async_version():
    tasks = [asyncio.sleep(2) for _ in range(3)]
    await asyncio.gather(*tasks)

start = time.time()
asyncio.run(async_version())
print(f"Асинхронно: {time.time() - start:.2f}s")

Вывод:

• Используйте asyncio.gather() для 95% случаев (просто и надёжно)
• TaskGroup() для Python 3.11+ (современный стандарт)
• wait() если нужен точный контроль над завершением
• create_task() для управления отдельными задачами

Асинхронность — это один из самых мощных инструментов Python для высоконагруженных приложений!