asyncio로 개발하기¶

비동기 프로그래밍은 고전적인 “순차적” 프로그래밍과 다릅니다.

이 페이지는 흔한 실수와 함정을 나열하고, 이를 피하는 방법을 설명합니다.

디버그 모드¶

기본적으로 asyncio는 프로덕션 모드로 실행됩니다. 개발을 쉽게 하려고 asyncio에는 디버그 모드를 제공합니다.

여러 가지 방법으로 asyncio 디버그 모드를 활성화할 수 있습니다:

PYTHONASYNCIODEBUG 환경 변수를 1로 설정.
-X dev 파이썬 명령 줄 옵션 사용.
debug=True를 asyncio.run()로 전달.
loop.set_debug()를 호출.

디버그 모드를 활성화하는 것 외에도, 다음을 고려하십시오:

asyncio 로거의 로그 수준을 logging.DEBUG로 설정, 예를 들어 응용 프로그램 시작 시 다음 코드 조각을 실행할 수 있습니다:
```
logging.basicConfig(level=logging.DEBUG)
```
ResourceWarning 경고를 표시하도록 warnings 모듈을 구성. 이렇게 하는 한 가지 방법은 -W default 명령 줄 옵션을 사용하는 것입니다.

디버그 모드가 활성화되면:

asyncio는 기다리지 않은 코루틴을 검사하고 로그 합니다; 이것은 “잊힌 await” 함정을 완화합니다.
많은 스레드 안전하지 않은 asyncio API(loop.call_soon()과 loop.call_at() 메서드와 같은)가 잘못된 스레드에서 호출될 때 예외를 발생시킵니다.
I/O 선택기의 실행 시간은 I/O 연산 수행에 너무 오래 걸리면 로그 됩니다.
100ms보다 오래 걸리는 콜백이 로그 됩니다. loop.slow_callback_duration 어트리뷰트는 “느린” 것으로 간주할 최소 실행 시간(초)을 설정하는 데 사용될 수 있습니다.

동시성과 다중 스레드¶

이벤트 루프는 스레드(일반적으로 주 스레드)에서 실행되며 그 스레드에서 모든 콜백과 태스크를 실행합니다. 태스크가 이벤트 루프에서 실행되는 동안, 다른 태스크는 같은 스레드에서 실행될 수 없습니다. 태스크가 await 표현식을 실행하면, 실행 중인 태스크가 일시 중지되고 이벤트 루프는 다음 태스크를 실행합니다.

다른 OS 스레드에서 콜백을 예약하려면, loop.call_soon_threadsafe() 메서드를 사용해야 합니다. 예:

loop.call_soon_threadsafe(callback, *args)

거의 모든 asyncio 객체는 스레드 안전하지 않습니다. 태스크나 콜백 외부에서 작동하는 코드가 없으면 일반적으로 문제가 되지 않습니다. 그러한 코드가 저수준 asyncio API를 호출해야 하면, loop.call_soon_threadsafe() 메서드를 사용해야 합니다, 예를 들어:

loop.call_soon_threadsafe(fut.cancel)

다른 OS 스레드에서 코루틴 객체를 예약하려면, run_coroutine_threadsafe() 함수를 사용해야 합니다. 결과에 액세스할 수 있도록 concurrent.futures.Future를 반환합니다:

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

# Later in another OS thread:

future = asyncio.run_coroutine_threadsafe(coro_func(), loop)
# Wait for the result:
result = future.result()

시그널을 처리하고 자식 프로세스를 실행하려면, 이벤트 루프를 메인 스레드에서 실행해야 합니다.

loop.run_in_executor() 메서드는 concurrent.futures.ThreadPoolExecutor와 함께 사용되어, 이벤트 루프가 실행되는 OS 스레드를 블록하지 않고 다른 OS 스레드에서 블로킹 코드를 실행할 수 있습니다.

블로킹 코드 실행하기¶

블로킹 (CPU 병목) 코드는 직접 호출하면 안 됩니다. 예를 들어, 함수가 CPU 집약적인 계산을 1초 동안 수행하면, 모든 동시 asyncio 태스크와 IO 연산이 1초 지연됩니다.

An executor can be used to run a task in a different thread or even in a different process to avoid blocking the OS thread with the event loop. See the loop.run_in_executor() method for more details.

로깅¶

asyncio는 logging 모듈을 사용하고, 모든 로깅은 "asyncio" 로거를 통해 수행됩니다.

기본 로그 수준은 logging.INFO며, 쉽게 조정할 수 있습니다:

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

await 하지 않은 코루틴 감지¶

코루틴 함수가 호출되었지만 기다리지 않을 때(예를 들어, await coro() 대신 coro())나 코루틴이 asyncio.create_task()로 예약되지 않으면 asyncio가 RuntimeWarning을 방출합니다:

import asyncio

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

async def main():
    test()

asyncio.run(main())

출력:

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

디버그 모드에서의 출력:

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()

일반적인 수정법은 코루틴을 await 하거나 asyncio.create_task() 함수를 호출하는 것입니다:

async def main():
    await test()

전달되지 않은 예외 감지¶

Future.set_exception()가 호출되었지만, Future 객체가 await 되지 않으면, 예외는 절대로 사용자 코드로 전파되지 않습니다. 이럴 때, Future 객체가 가비지 수집될 때 asyncio가 로그 메시지를 출력합니다.

처리되지 않은 예외의 예:

import asyncio

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

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

asyncio.run(main())

출력:

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

태스크가 만들어진 곳의 트레이스백을 얻으려면 디버그 모드를 활성화하세요:

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