asyncio 비동기 처리와 장점

Python은 asyncio 라이브러리를 활용해 비동기 실행을 지원한다. 하지만 asyncio의 경우, 파이썬 버전에 따라 많은 변화가 있었다. 아래 글에 포함된 코드는 Python 3.9.12를 활용해 코드를 실행해 보았다. 특히 3.7 이전의 버전을 활용한다면 아래 글의 예제 코드가 실행되지 않을 수 있다.

import asyncio

동기와 비동기 방식을 간략하게 표현하자면 위 그림과 같다. main 작업이 진행되는 동안 동시에(concurrent) 다른 작업이 진행될 수 있는 것이다. (단, 동시에 처리되는 것이지 병렬적으로 처리되는 것은 아니다.)

코루틴 선언

async은 네이티브 코루틴을 선언하는 방식이다. def로 함수를 선언하는 것과 문법이 동일하지만 def 앞에 async을 붙여 사용한다.

async def co():

이렇게 생성된 네이티브 코루틴 내부에는 await 키워드를 사용할 수 있다. await는 코루틴의 결과를 기다린다. 만약 네이티브 코루틴을 호출하더라도 await가 되지 않으면 코루틴 객체를 반환할 뿐 실행하지 않는다.

async def f():
    return 30

async def main():
    ret = f()
    # ret: <coroutine object f at ...>
    # 코루틴 객체 생성되었지만 await하고 있지 않기 때문에
    # 아무런 일도 발생하지 않는다.

    ret = await f()
    # ret: 30  
    # 실제 코루틴이 실행되고 값을 반환한다.

코루틴 실행

asyncio.run(코루틴)

run은 코루틴을 실행한다. 일반적으로 아래와 같이 main 코루틴을 하나 생성해 run에 전달한다.

async def main():
    ...

asyncio.run(main())

run은 3.7에 새로 생긴 기능이기 때문에 이전 버전은 아래와 같이 실행해야 한다.

loop = asyncio.get_event_loop()
loop.run_until_complete(main())  # main은 코루틴 객체이다.
loop.close()

또한 event loop에서 저수준(low-level)의 작업을 하기 위해서는 위와 같이 loop을 명시적으로 가져와 사용하는 방법을 사용한다.

create_task

코루틴을 생성하고 독립적인 태스크로 실행되기 위해서는 create_task를 사용해야 한다. 이 함수도 3.7에 처음 추가되었다.

asyncio.create_task(코루틴)

asyncio에는 Coroutine, Task, Feature 객체가 존재한다. 그리고 3종류의 객체를 Awaitable 객체라고 부른다. 따라서 create_task는 정확히 말하면 Coroutine을 입력받아 Task 객체로 반환하는 역할을 하는 함수이다. Task 객체가 생성되고 나면 코루틴은 바로 실행될 수 있도록 예약된다. 공식 문서 "task-object"를 한 번 읽어보는 것을 추천한다.

async def main():
    task = asyncio.create_task(f())
    # 코루틴이 task로 예약되고 실행되었다.
    # 이제 task의 실행 결과를 기다릴 수 있다.
    await task

아래 함수를 실행해보면 동시에 비동기로 실행되는 것을 볼 수 있다.

import asyncio
import time

async def f(t):
    """ 실행에 약 t초가 소요되는 함수 """
    await asyncio.sleep(t)

async def main():
    task1 = asyncio.create_task(f(6))
    task2 = asyncio.create_task(f(7))

    await task1
    await task2


start = time.time()

ret = asyncio.run(main())

end = time.time()
print(f"시간: {round(end - start)}초")
# 시간: 7초

동기로 실행되었다면 6초가 걸리는 함수와 7초가 걸리는 함수가 순서대로 실행되어 약 13초가 소요되었을 것이다. 하지만 위 코드는 약 7초가 소요되었다. 이를 통해 우리가 의도한 대로 실행된다는 것을 알 수 있다.

* asyncio.sleep는 time.sleep과 같은 역할을 하는 non-blocking 함수이다. time.sleep은 blocking 함수이기 때문에 time.sleep을 활용해 실행해보면 약 13초가 소요된다.

gather

gather는 번거롭게 하나씩 await할 필요 없이 여러 task의 반환 값을 기다릴 수 있다. task를 통해 반환된 값들은 리스트 형태로 반환된다.

await asyncio.gather(task1, task2, ... )
-> [task1 반환값, task2 반환값, ... ]

아래 예시와 같이 사용할 수 있다.

import asyncio

async def sqaure(s):
    return s ** 2

async def main():
    tasks = [sqaure(x) for x in [2, 3, 7]]
    ret = await asyncio.gather(*tasks)
    return ret

ret = asyncio.run(main())
print(f"결과: {ret}")
# 결과: [4, 9, 49]

비동기 + Non-blocking

Python의 함수는 기본적으로 Blocking 상태이다. 함수가 실행되는 동안 main 함수는 아무것도 하지 못하고 반환 값을 기다려야만 한다. 하지만 event loop에 run_in_executor를 활용하면 Non-blocking으로 동작할 수 있다.

# loop = asyncio.get_event_loop()
loop.run_in_executor(None, 함수, 인자1, 인자2 ... )

예시: (requests로 HTML 가져오기)

import asyncio
import time
import requests

urls = ["https://www. ... ", ... ]  # 10개의 url 주소
headers = { "User-Agent": "Mozilla/5.0 ... "}

async def get_reqeust(url):
    request = await loop.run_in_executor(None, requests.get, url, headers)
    return request.status_code

async def main():
    tasks = [asyncio.create_task(get_reqeust(url)) for url in urls]
    ret = await asyncio.gather(*tasks)
    return ret


start = time.time()

loop = asyncio.get_event_loop()
status = loop.run_until_complete(main())
loop.close()

end = time.time()
print(f"시간: {round(end - start)}초, 실행 결과: {status}")

시간: 1초, 실행 결과: [200, 200, 200, 200, 200, 200, 200, 200, 200, 200]

10개의 URL에 접속해 모두 정상적(200)으로 정보를 가져왔으며 총 1초가 소요되었다.

같은 작업을 동기 방식으로 진행해 보았다.

# 생략

def get_reqeust(url):
    request = requests.get(url, headers)
    return request.status_code

def main():
    ret = [get_reqeust(url) for url in urls]
    return ret

start = time.time()

status = main()

end = time.time()
print(f"시간: {round(end - start)}초, 실행 결과: {status}")

시간: 7초, 실행 결과: [200, 200, 200, 200, 200, 200, 200, 200, 200, 200]

같은 URL에 접속해 정보를 가져왔지만 시간이 7배 정도 더 오래 걸렸다. 접속해야 하는 URL의 수가 많을수록 차이는 더 심해질 것이다.

위 결과를 정리해보면 아래 표와 같다.

	동기	비동기
접속한 URL	URL 10개
소요 시간	약 7초	약 1초

async with / for

기존의 컨텍스트 매니저, 이터레이터와 상당히 유사하다.

자세한 내용은 아래 더보기 참고

with

class ConText():
    async def __aenter__(self):
        ...

    async def __aexit__(self, exc_type, exc_value, traceback):
        ...


async def f():
    async with ConText() as context:
        ...

for

class Iterator():
    def __aiter__(self):
        ...

    async def __anext__(self):
        ...
        # StopIteration 대신 StopAsyncIteration 발생


async def f():
    async for i in Iterator():
        ...

Python 비동기 장점

"비동기 + Non-blocking"에서도 확인했듯이 Request, I/O bound 프로세스와 같이 딜레이가 발생하는 작업에서 뛰어난 효과를 보인다. 쉽게 말해 서버에 정보를 요청하거나 데이터를 읽는 등 작업에 유리하다는 의미이다. 대기 시간이 발생하는데 상황에서 이러한 시간을 다른 작업을 수행하는데 활용함으로써 전체적인 소요 시간이 감소하는 것이다.

하지만 CPU bound 작업은 다르다. 파이썬의 비동기는 병렬적(parallel)으로 처리되는 것이 아니라 동시(concurrent)에 처리되는 것이다 (GIL 참고). 반복적인 연산을 수행하는 코드를 실행해보면 오히려 비동기로 처리했을 때 더 많은 시간이 소요되었다. 일반 이터레이터(동기 방식)을 사용했을 때 21초가 소요되는 작업을 비동기 이터레이터는 29초가 소요되었다.

사람 한 명이 커피 포트에 물을 올려놓고 물이 끓는 동안 잠시 피아노 연습을 하는 것은 시간을 아껴준다. 그런데 커피콩을 갈면서 동시에 피아노 연습을 하는 것은 오히려 시간이 더 오래 걸리는 비효율적인 행동이다. I/O 작업은 대기 시간이 있기 때문에 남는 시간 동안 다른 작업을 처리할 수 있지만 CPU가 바쁘게 움직이고 있는데 동시에 다른 작업을 처리하도록 하는 것은 비효율적이다.

따라서 파이썬에서 비동기 작업을 할 때는 해결하고자 하는 문제가 무엇인지 확인하고 적절한 방식을 선택하는 것이 중요하다.