11.01.2026 07:40
Structured Concurrency в Python 3.11+: TaskGroup vs Gather
Structured Concurrency в Python 3.11+ (TaskGroup vs Gather)
Если вы до сих пор используете asyncio.gather() для запуска конкурентных задач, вы, вероятно, теряете контроль над ошибками.
Главная проблема gather: если одна таска падает, остальные продолжают работать (если не стоит return_exceptions=False, который убивает всё сразу, но "грязно"). Плюс, отлавливать несколько ошибок одновременно через try / except - это боль.
В Python 3.11+ (и через trio / anyio раньше) стандартом стала Structured Concurrency через asyncio.TaskGroup.
Как это выглядит:
import asyncio
async def task_fail(name, sec):
await asyncio.sleep(sec)
raise ValueError(f"Error in {name}")
async def task_ok(name, sec):
await asyncio.sleep(sec)
print(f"Task {name} done")
async def main():
try:
async with asyncio.TaskGroup() as tg:
tg.create_task(task_ok("A", 1))
tg.create_task(task_fail("B", 2)) # Упадет через 2 сек
tg.create_task(task_fail("C", 1.5)) # Упадет через 1.5 сек
except* ValueError as eg:
# Обратите внимание на except* (ExceptionGroup)
print(f"Caught errors: {eg.exceptions}")
# Запуск
# asyncio.run(main())
Почему TaskGroup это выбор Middle+:
1. Атомарность скоупа: Если одна задача внутри контекстного менеджера падает, TaskGroup автоматически отменяет (cancel) все остальные запущенные задачи в группе. Вы не оставляете "зомби-процессы" в фоне.
2. Exception Groups: Если упало несколько задач одновременно (или во время отмены), Python соберет их в ExceptionGroup. Конструкция except* позволяет элегантно обрабатывать дерево исключений.
Вывод:
Для скриптов gather сойдет. Для надежных сервисов, где важна консистентность состояния и корректный шатдаун корутин - переезжайте на TaskGroup.
#asyncio #python311 #bestpractices #concurrency
👉 @BookPython
Если вы до сих пор используете asyncio.gather() для запуска конкурентных задач, вы, вероятно, теряете контроль над ошибками.
Главная проблема gather: если одна таска падает, остальные продолжают работать (если не стоит return_exceptions=False, который убивает всё сразу, но "грязно"). Плюс, отлавливать несколько ошибок одновременно через try / except - это боль.
В Python 3.11+ (и через trio / anyio раньше) стандартом стала Structured Concurrency через asyncio.TaskGroup.
Как это выглядит:
import asyncio
async def task_fail(name, sec):
await asyncio.sleep(sec)
raise ValueError(f"Error in {name}")
async def task_ok(name, sec):
await asyncio.sleep(sec)
print(f"Task {name} done")
async def main():
try:
async with asyncio.TaskGroup() as tg:
tg.create_task(task_ok("A", 1))
tg.create_task(task_fail("B", 2)) # Упадет через 2 сек
tg.create_task(task_fail("C", 1.5)) # Упадет через 1.5 сек
except* ValueError as eg:
# Обратите внимание на except* (ExceptionGroup)
print(f"Caught errors: {eg.exceptions}")
# Запуск
# asyncio.run(main())
Почему TaskGroup это выбор Middle+:
1. Атомарность скоупа: Если одна задача внутри контекстного менеджера падает, TaskGroup автоматически отменяет (cancel) все остальные запущенные задачи в группе. Вы не оставляете "зомби-процессы" в фоне.
2. Exception Groups: Если упало несколько задач одновременно (или во время отмены), Python соберет их в ExceptionGroup. Конструкция except* позволяет элегантно обрабатывать дерево исключений.
Вывод:
Для скриптов gather сойдет. Для надежных сервисов, где важна консистентность состояния и корректный шатдаун корутин - переезжайте на TaskGroup.
#asyncio #python311 #bestpractices #concurrency
👉 @BookPython