> **Источник:** https://python-all.ru/3.7/library/asyncio-dev.html
>
> «Документация Python на русском» – неофициальный перевод официальной документации Python: версии от 2.6 до 3.16, полнотекстовый поиск, английский оригинал рядом с переводом. Эта Markdown-версия страницы предназначена для работы с LLM: вставьте её в ChatGPT, Claude или Cursor.

---

# Разработка с asyncio

Асинхронное программирование отличается от классического «последовательного» программирования.

На этой странице перечислены типичные ошибки и ловушки, а также объясняется, как их избежать.

## Режим отладки

По умолчанию asyncio работает в производственном режиме. Для упрощения разработки в asyncio есть *режим отладки*.

Включить режим отладки asyncio можно несколькими способами:

- Установка переменной окружения [`PYTHONASYNCIODEBUG`](https://python-all.ru/3.7/using/cmdline.html#envvar-PYTHONASYNCIODEBUG) в значение `1`.
- Использование опции командной строки Python [`-X`](https://python-all.ru/3.7/using/cmdline.html#id5) `dev`
- Передача `debug=True` в [`asyncio.run()`](https://python-all.ru/3.7/library/asyncio-task.html#asyncio.run).
- Вызов [`loop.set_debug()`](https://python-all.ru/3.7/library/asyncio-eventloop.html#asyncio.loop.set_debug).

В дополнение к включению режима отладки также рекомендуется:

- Установка уровня журналирования [логгера asyncio](https://python-all.ru/3.7/library/asyncio-dev.html#asyncio-logger) на `logging.DEBUG`; например, следующий фрагмент кода можно выполнить при запуске приложения:

  ```python
  logging.basicConfig(level=logging.DEBUG)
  ```
- Настройка модуля [`warnings`](https://python-all.ru/3.7/library/warnings.html#module-warnings) на отображение предупреждений [`ResourceWarning`](https://python-all.ru/3.7/library/exceptions.html#ResourceWarning). Один из способов сделать это – использовать параметр командной строки [`-W`](https://python-all.ru/3.7/using/cmdline.html#cmdoption-w) `default`.

Когда режим отладки включён:

- asyncio проверяет [корутины, которые не были ожиданы](https://python-all.ru/3.7/library/asyncio-dev.html#asyncio-coroutine-not-scheduled) и регистрирует их; это помогает избежать ошибки «забытого await».
- Многие небезопасные по отношению к потокам API asyncio (например, методы [`loop.call_soon()`](https://python-all.ru/3.7/library/asyncio-eventloop.html#asyncio.loop.call_soon) и [`loop.call_at()`](https://python-all.ru/3.7/library/asyncio-eventloop.html#asyncio.loop.call_at)) вызывают исключение, если их вызывают не из того потока.
- Время выполнения селектора ввода-вывода журналируется, если выполнение операции ввода-вывода занимает слишком много времени.
- Колбэки, выполняющиеся дольше 100 мс, записываются в журнал. С помощью атрибута `loop.slow_callback_duration` можно задать минимальную длительность выполнения в секундах, которая считается «медленной».

## Параллелизм и многопоточность

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

Чтобы запланировать колбэк из другого потока ОС, следует использовать метод [`loop.call_soon_threadsafe()`](https://python-all.ru/3.7/library/asyncio-eventloop.html#asyncio.loop.call_soon_threadsafe). Пример:

```python
loop.call_soon_threadsafe(callback, *args)
```

Почти все объекты asyncio не являются потокобезопасными, что обычно не является проблемой, если только нет кода, который работает с ними вне задачи или колбэка. Если такому коду нужно вызвать низкоуровневый API asyncio, следует использовать метод [`loop.call_soon_threadsafe()`](https://python-all.ru/3.7/library/asyncio-eventloop.html#asyncio.loop.call_soon_threadsafe), например:

```python
loop.call_soon_threadsafe(fut.cancel)
```

Для планирования объекта корутины из другого потока ОС следует использовать функцию [`run_coroutine_threadsafe()`](https://python-all.ru/3.7/library/asyncio-task.html#asyncio.run_coroutine_threadsafe). Она возвращает [`concurrent.futures.Future`](https://python-all.ru/3.7/library/concurrent.futures.html#concurrent.futures.Future) для доступа к результату:

```python
async def coro_func():
     return await asyncio.sleep(1, 42)

# Позже в другом потоке ОС:

future = asyncio.run_coroutine_threadsafe(coro_func(), loop)
# Ожидать результат:
result = future.result()
```

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

Метод [`loop.run_in_executor()`](https://python-all.ru/3.7/library/asyncio-eventloop.html#asyncio.loop.run_in_executor) можно использовать с [`concurrent.futures.ThreadPoolExecutor`](https://python-all.ru/3.7/library/concurrent.futures.html#concurrent.futures.ThreadPoolExecutor) для выполнения блокирующего кода в другом потоке ОС, не блокируя поток ОС, в котором работает цикл событий.

## Выполнение блокирующего кода

Блокирующий (связанный с процессором) код не следует вызывать напрямую. Например, если функция выполняет интенсивные вычисления в течение 1 секунды, все параллельные задачи asyncio и операции ввода-вывода будут задержаны на 1 секунду.

Исполнитель можно использовать для выполнения задачи в другом потоке или даже в другом процессе, чтобы избежать блокировки потока ОС циклом событий. Подробнее см. метод [`loop.run_in_executor()`](https://python-all.ru/3.7/library/asyncio-eventloop.html#asyncio.loop.run_in_executor) для получения дополнительных сведений.

## Журналирование

asyncio использует модуль [`logging`](https://python-all.ru/3.7/library/logging.html#module-logging), а все логирование выполняется через логгер `"asyncio"`.

Уровень логирования по умолчанию – `logging.INFO`. Его можно легко изменить:

```python
logging.getLogger("asyncio").setLevel(logging.WARNING)
```

## Обнаружение никогда не ожидаемых корутин

Когда корутинная функция вызывается, но не ожидается (например, `coro()` вместо `await coro()`) или корутина не запланирована с помощью [`asyncio.create_task()`](https://python-all.ru/3.7/library/asyncio-task.html#asyncio.create_task), asyncio выдаёт сообщение [`RuntimeWarning`](https://python-all.ru/3.7/library/exceptions.html#RuntimeWarning):

```python
import asyncio

async def test():
    print("never scheduled")

async def main():
    test()

asyncio.run(main())
```

Вывод:

```python
test.py:7: RuntimeWarning: coroutine 'test' was never awaited
  test()
```

Вывод в режиме отладки:

```python
test.py:7: RuntimeWarning: coroutine 'test' was never awaited
Coroutine created at (most recent call last)
  File "../t.py", line 9, in <module>
    asyncio.run(main(), debug=True)

  < .. >

  File "../t.py", line 7, in main
    test()
  test()
```

Обычно это исправляется либо ожиданием корутины, либо вызовом функции [`asyncio.create_task()`](https://python-all.ru/3.7/library/asyncio-task.html#asyncio.create_task):

```python
async def main():
    await test()
```

## Обнаружение никогда не извлечённых исключений

Если вызван метод [`Future.set_exception()`](https://python-all.ru/3.7/library/asyncio-future.html#asyncio.Future.set_exception), но сам объект Future никогда не ожидается, исключение не будет передано пользовательскому коду. В этом случае asyncio выдаёт сообщение в журнал при сборке мусора для этого Future.

Пример необработанного исключения:

```python
import asyncio

async def bug():
    raise Exception("not consumed")

async def main():
    asyncio.create_task(bug())

asyncio.run(main())
```

Вывод:

```python
Task exception was never retrieved
future: <Task finished coro=<bug() done, defined at test.py:3>
  exception=Exception('not consumed')>

Traceback (most recent call last):
  File "test.py", line 4, in bug
    raise Exception("not consumed")
Exception: not consumed
```

[Включите режим отладки](https://python-all.ru/3.7/library/asyncio-dev.html#asyncio-debug-mode), чтобы получить трассировку стека на момент создания задачи:

```python
asyncio.run(main(), debug=True)
```

Вывод в режиме отладки:

```python
Task exception was never retrieved
future: <Task finished coro=<bug() done, defined at test.py:3>
    exception=Exception('not consumed') created at asyncio/tasks.py:321>

source_traceback: Object created at (most recent call last):
  File "../t.py", line 9, in <module>
    asyncio.run(main(), debug=True)

< .. >

Traceback (most recent call last):
  File "../t.py", line 4, in bug
    raise Exception("not consumed")
Exception: not consumed
```