asyncio 비동기 프로그래밍 (250710)

asyncio 비동기 프로그래밍

MCP 서버 관련 학습을 하는 중 asyncio 비동기 프로그래밍 부분에 대한 내용이 있어 정리해본다.

# asyncio 비동기 프로그래밍...

웹 서버나 게임 서버에서 빠른 응답 시간을 요구하는 경우 asyncio를 사용하면, 
수천 개의 클라이언트 요청을 동시에 처리할 수 있다. 

서버는 비동기 방식으로 각 클라이언트 요청을 처리하고, 
요청이 끝날 때까지 기다리는 동안 다른 요청을 처리할 수 있다.

간단한 예제를 통해 실습을 하는 중에 12세대 인텔과 Mac M1에서 실행한 결과에 차이가 있어서 간단히 기록으로 남긴다.

print()는 표준 출력 버퍼를 통해 I/O를 수행하므로 오히려 asyncio 환경에서는 더 느릴 수 있다. Mac M1에서 실행할 경우에도 CPU 성능이 워낙 뛰어나기 때문에 두 코드가 비슷한 속도로 수행되며, 오히려 asyncio 코드의 오버헤드 때문에 아주 약간 더 느릴 수도 있다.

동기 처리 예제

from time import time

def add_a(a, b):
    for i in range(300000):  # 반목문에 의한 임의의 시간 지연 생성
        print(i)


    print('add_a: {0} + {1}'.format(a, b))
    
    return a+b

def add_b(a, b):
    for i in range(600000):  # 반복문에 의한 시간 지연 생성 - add_a 보다 약간 길게
        print(i)

    print('add_b: {0} + {1}'.format(a, b))

    return a+b

def print_add(a, b):
    add_a(a, b)            
    result = add_b(a, b)   
    print('print_add_b: {0} + {1} = {2}'.format(a, b, result))


start = time()

# 임의의 반복문을 설정하여 시간 지연을 생성함
print_add(1, 2)

end = time()

print('실행 시간 : {0:.5f}초'.format(end-start))

동기 처리 실행결과

...
599998
599999
add_b: 1 + 2
print_add_b: 1 + 2 = 3
실행 시간 : 1.57579초

비동기 처리 예제

import asyncio
from time import time

async def add_a(a, b):
    for i in range(300000):  # 반목문에 의한 임의의 시간 지연 생성
        print(i)


    print('add_a: {0} + {1}'.format(a, b))
    
    return a+b

async def add_b(a, b):
    for i in range(600000):  # 반복문에 의한 시간 지연 생성 - add_a 보다 약간 길게
        print(i)

    print('add_b: {0} + {1}'.format(a, b))

    return a+b

async def print_add(a, b):
    await add_a(a, b)            # await로 다른 네이티브 코루틴 실행
    result = await add_b(a, b)   # await로 다른 네이티브 코루틴 실행하고 반환 값을 변수에 저장
    print('print_add_b: {0} + {1} = {2}'.format(a, b, result))


start = time()

# 이벤트 루프를 얻는 과정까지 소모되는 시간을 초과하도록
# 임의의 반복문을 설정하여 시간 지연을 생성함

loop = asyncio.get_event_loop()           # 이벤트 루프를 획득
loop.run_until_complete(print_add(1, 2))  # print_add(1, 2)가 끝날 때까지 이벤트 루프를 실행
loop.close()                              # 이벤트 루프를 닫음

end = time()

print('실행 시간 : {0:.5f}초'.format(end-start))

비동기 처리 실행 결과

...
599997
599998
599999
add_b: 1 + 2
print_add_b: 1 + 2 = 3
실행 시간 : 1.60377초

asyncio.sleep()을 이용하여 테스트

실제로 asyncio의 성능을 테스트하고 싶다면, 다음과 같이 I/O 지연이나 asyncio.sleep()을 사용하여 테스트해야 의미 있는 결과를 얻을 수 있다.

동기 처리 예제

from time import sleep  # time 모듈에서 sleep 함수만 임포트
import time

def add_a(a, b):
    sleep(1)  # I/O 작업 시뮬레이션
    return a + b

def add_b(a, b):
    sleep(2)  # 더 긴 I/O 작업 시뮬레이션
    return a + b

def main():
    start = time.time()  # time 모듈의 time() 함수 호출
    result_a = add_a(1, 2)  # 순차적 실행
    result_b = add_b(1, 2)  # 순차적 실행
    print("동기 실행 결과:", result_a, result_b)  # 결과값 출력
    print("동기 실행 시간:", time.time() - start)

main()

동기 처리 예제 실행 결과

동기 실행 결과: 3 3
동기 실행 시간: 3.0100338459014893

비동기 처리 예제

import asyncio
from time import time

async def add_a(a, b):
    await asyncio.sleep(1)  # simulate I/O
    return a + b

async def add_b(a, b):
    await asyncio.sleep(2)  # longer I/O
    return a + b

async def main():
    start = time()
    await asyncio.gather(add_a(1, 2), add_b(1, 2))  # 비동기 병렬 실행
    print("Async 실행 시간:", time() - start)

asyncio.run(main())

비동기 처리 예제 실행 결과

Async 실행 시간: 2.003904104232788