Python에서 비동기 프로그래밍

코루틴(Coroutine)

코루틴은 함수 실행의 지연과 재개를 허용하는 함수이다. 여러 작업을 동시에 처리하는 비동기 프로그래밍에 유용하다.

파이썬에서 코루틴은 일반적으로 async/await 문법이 선호된다. async 키워드를 사용하여 코루틴을 정의하며, await를 통해 다른 작업이 완료될 때까지 대기할 수 있다.

 

기본 사용법

asyncio 라이브러리를 사용해서 구현한다.

함수를 선언할때, def 앞에 async 를 추가해서 코루틴을 정의하며, 코루틴을 단순히 호출하는것이 아닌 aysncio.run() 을 이용해서 실행한다.

import asyncio

async def async_function(name, delay):
    print(f"Start {name}")
    await asyncio.sleep(delay)
    print(f"End {name}")

async def main():
    await async_function("Task 1", 2)

if __name__ == "__main__":
    asyncio.run(main())

'''
출력 결과:

Start Task 1
End Task 1
'''

 

동시에 실행하기

위의 예제와 달리 여러개의 태스크를 동시에 실행하고자 한다면, 아래의 방법들을 활용 할 수 있다.

세개의 태스크가 동시에 실행되면서, 지연시간이 가장 짧은 Task2가 먼저 End되는것을 확인 할 수 있다.

• asyncio.create_task()

import asyncio


async def async_function(name, delay):
    print(f"Start {name}")
    await asyncio.sleep(delay)
    print(f"End {name}")


async def main():
    task1 = asyncio.create_task(
        async_function("Task 1", 2))
    task2 = asyncio.create_task(
        async_function("Task 2", 1))
    task3 = asyncio.create_task(
        async_function("Task 3", 3))

    await task1
    await task2
    await task3


if __name__ == "__main__":
    asyncio.run(main())

'''
출력 결과:

Start Task 1
Start Task 2
Start Task 3
End Task 2
End Task 1
End Task 3
'''

• asyncio.TaskGroup()

import asyncio


async def async_function(name, delay):
    print(f"Start {name}")
    await asyncio.sleep(delay)
    print(f"End {name}")


async def main():
    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(
            async_function("Task 1", 2))
        task2 = tg.create_task(
            async_function("Task 2", 1))
        task3 = tg.create_task(
            async_function("Task 3", 3))


if __name__ == "__main__":
    asyncio.run(main())

'''
출력 결과:

Start Task 1
Start Task 2
Start Task 3
End Task 2
End Task 1
End Task 3
'''

• asyncio.gather()
  어웨이터블 객체(Future, task, 코루틴)를 받아서 동시에 실행

import asyncio

async def async_function(name, delay):
    print(f"Start {name}")
    await asyncio.sleep(delay)
    print(f"End {name}")

async def main():
	await asyncio.gather(
	        async_function("Task 1", 2),
	        async_function("Task 2", 1),
	        async_function("Task 3", 3)
	    )

if __name__ == "__main__":
    asyncio.run(main())

'''
출력 결과:

Start Task 1
Start Task 2
Start Task 3
End Task 2
End Task 1
End Task 3
'''

 

FastAPI에서 활용

FastAPI에서 비동기로 API를 구현할때도 asyncio를 활용 할 수 있다.

아래의 예제는 API가 호출되면 중간 task의 실행을 기다리지 않고 바로 return값을 반환하도록 구현했다.

import asyncio

import uvicorn
from fastapi import FastAPI


app = FastAPI()


async def async_function():
    await asyncio.sleep(2)
    print("async_function execute")


@app.get("/async/test")
async def async_test():
    asyncio.create_task(async_function())
    ret = {"code": 200, "msg": "성공"}
    print(ret)
    return ret


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)
    
'''
출력 결과:

{'code': 200, 'msg': '성공'}
INFO:     127.0.0.1:56741 - "GET /async/test HTTP/1.1" 200 OK
async_function execute
'''