비동기 프로그래밍 2편 - 간단한 예제

안녕하세요, 이번 시간엔 저번에 이어 비동기 프로그래밍이 어떻게 사용되는지 간단한 예제를 통해 이해해보려고 해요.

 

여기서 진행한 프로젝트 코드는 제 깃허브에 들어가시면 볼 수 있어요.

GitHub - dalmengs/async-test-blog

---

1. 파이썬에서의 비동기 프로그래밍

2. 비동기 프로그래밍 예제

3. 비동기 프로그래밍 활용 1 - GPT API 비동기 처리하기

4. 비동기 프로그래밍 활용 2 - 복잡한 회원가입 로직 처리하기

---

2. 비동기 프로그래밍 예제 - Asynchronous Programming Example

오늘의 목표는 1초가 걸리는 작업 10개를 처리하는 거예요.

아래 두 함수는 같은 동작을 하고, 실행에 1초가 걸리는 간단한 함수예요.

def sync_api(n: int):
    time.sleep(1)
    print(f"sync_api {n} Finished")


async def async_api(n: int):
    await asyncio.sleep(1)
    print(f"async_api {n} Finished")

 

sync_api 함수는 동기 함수이고, async_api는 비동기 방식의 함수예요.

 

2 - 1. 동기 방식으로 처리하기

가장 처리하기 쉬운 방식은 역시 가장 익숙한 동기 방식으로 처리하는 거죠.

반복문을 이용하여 10번 반복하여 sync_api 함수를 호출해주면 쉽게 목표를 달성할 수 있어요.

@AsyncExecutionTime("Main")
async def main1():
    for i in range(10):
        sync_api(i)

 

동기 방식으로 처리했기 때문에 sync_api가 실행되는 동안 다른 어떤 동작도 할 수 없어요.

따라서 실행 시간이 10초가 나온 것을 볼 수 있죠.

 

목표는 달성했지만 이런 쉬운 작업을 하는 데 10초가 기다려야 하다니, 맘에 들지 않네요.

이전 게시물에서 소개한 비동기 프로그래밍을 활용해서 더 효율적으로 만들어봅시다.

 

2 - 2. 비동기 방식으로 처리하기

sync_api 함수 대신 async_api 함수를 호출해서 비동기적으로 처리해봅시다.

@AsyncExecutionTime("Main")
async def main2():
    for i in range(10):
        await async_api(i)

 

이로써 각 함수를 비동기적으로 수행할 수 있게 되었어요.

 

하지만 결과를 보니 똑같이 10초가 걸렸네요. 왜 그런 결과가 나온 걸까요?

그 이유는 비동기 함수의 결과를 await으로 기다렸기 때문이에요.

 

비동기 함수를 호출했기 때문에 동시에 다른 로직을 수행할 수 있지만 await 키워드로 인해 결과가 반환될 때까지 기다려야 하기 때문에 다른 비동기 함수가 수행되지 못 한거죠.

 

2 - 3. create_task 함수 활용하기

지금 문제는 비동기 함수의 결과를 쓸데없이 기다려야 한다는 거죠.

그러면 이전 시간에 소개한 create_task 함수를 활용해서 효율적으로 만들어볼까요?

 

결과를 기다리는 것은 좋지만, 모든 10개의 비동기 함수를 모두 실행한 뒤 결과를 기다리면 더 좋지 않을까요?

@AsyncExecutionTime("Main")
async def main3():
    tasks = []
    for i in range(10):
        tasks.append(asyncio.create_task(async_api(i)))
    await asyncio.gather(*tasks)

 

await으로 결과를 기다리지 않고 create_task로 비동기 함수를 모두 실행해줬어요.

그런 뒤에 asyncio 라이브러리의 gather 함수를 이용하여 실행한 태스크의 결과를 한 번에 기다려줬어요.

 

이렇게 하면 동시에 10개의 비동기 함수가 실행되기 때문에 효율적으로 처리할 수 있겠죠?

 

결과도 1초가 나온 것을 알 수 있어요. 이로써 10초가 걸리던 작업을 1초로 줄일 수 있었어요.

 

2 - 4. create_task 함수를 활용한 다른 방법

gather 함수의 존재를 모른다면 아래와 같이 구현해도 돼요.

@AsyncExecutionTime("Main")
async def main4():
    tasks = []
    for i in range(10):
        tasks.append(asyncio.create_task(async_api(i)))
    for task in tasks:
        await task

 

한 번에 실행하고 모든 결과를 기다려야 한다면 gather 함수로도 충분해요.

하지만 이 방법이 가지는 큰 장점이 있죠.

 

동시에 비동기적으로 실행하되, 결과를 실행했던 순서대로 받을 수 있다는 점이에요.

 

챗봇이 하는 말을 음성으로 변환해서 사용자에게 들려준다고 가정해볼게요.

예를 들어, 챗봇이 ("배고프다", "밥을 먹었다") 라는 말을 했어요. 이해하면 배고파서 밥을 먹었다고 이해할 수 있겠죠?

 

이를 동기적으로 처리하면 아래 그림과 같아요.

딱 봐도 비효율적이라는 것이 느껴지죠? 첫 문장이 끝나기까지 기다린 후 두 번째 문장 처리를 할 수 있어요.

 

"2 - 3. create_task 함수 활용하기"와 동일한 방법으로 처리해본다고 가정해볼게요.

async def chat(messages):
    tasks = []
    for message in messages:
        tasks.append(asyncio.create_task(text_to_speech(message)))
    await asyncio.gather(*tasks)

 

이 처리의 흐름을 보면 아래 그림과 같아요.

"음성 변환"이 아니라 "결과 전달" 정도로 바꿔서 이해해주세요.

태스크가 완료되는 순서를 제어하지 않았기 때문에 뒷 문장 변환이 빨리돼서 사용자가 뒷 문장을 먼저 듣게 된 경우예요.

이 경우에는 ("밥을 먹었다", "배고프다")의 순서로 사용자에게 전달되죠. 의도는 배고파서 밥을 먹은 거지만 사용자는 밥을 먹었지만 배고프다고 이해할 수도 있겠죠?

 

동기적으로 처리하면 비효율적이라는 것을 위에서 봤어요. 근데 비동기적으로 위와 같이 처리하면 우리가 의도한 결과와 다를 수도 있다는 것을 봤어요.

따라서 비동기적으로 실행은 하되, 결과의 순서는 보장해야 돼요.

 

따라서 아래와 같이 구현하면 우리가 원하는 대로 구현할 수 있어요.

async def chat(messages):
    tasks = []
    for message in messages:
        tasks.append(asyncio.create_task(text_to_speech(message)))
    for task in tasks:
    	speech = await task
        send_to_client(speech)

 

위 코드의 실행 흐름을 그림으로 나타내면 아래와 같아요.

"음성 변환"이 아니라 "결과 전달" 정도로 바꿔서 이해해주세요.

이로써 우리가 의도한 대로 비동기적으로 처리해서 효율성은 높이되, 순서를 보장해서 우리가 원하는 결과를 낼 수 있게 되었어요.

---

아주 간단한 예시만 보았지만 비동기 프로그래밍을 활용하면 효율성을 크게 높일 수 있다는 것을 알 수 있겠죠?

이해하기 쉽지는 않지만, 잘 활용한다면 아주 효율적이고 좋은 프로그램을 만들 수 있어요.

 

다음 게시물에서는 GPT-3.5 API 를 활용해서 실제로 활용하는 간단한 프로젝트를 해볼게요.

---

제 글이 많은 도움이 되었길 바라며,
긴 글 봐주셔서 감사합니다!
멋진 개발자가 되기 위해 더 열심히 달리겠습니다!
- 달맹 -