🧐파이썬 코드 잘 짜는 법 : 비동기 처리로 성능 최적화 하기

 

파이썬 비동기 처리로 개선하다

 

코틀린에서 코루틴으로 비동기처리를 했는데, 파이썬에서도 비동기처리를 알게 되어서 정리해본다. 

예를 들어, LLM을 사용해서 이미지를 학습하려고 하는데 이때 수많은 이미지를 동시에 처리하는 작업이 필요했다.

이미지들을 다운로드 하고, 변환하고, 전처리하는 각각의 과정을 하나씩 순차적으로 처리한다면 시간이 너무 오래걸릴 것이다. 

그래서 파이썬의 비동기 프로그래밍을 활용해서 이러한 작업들이 동시에 실행되도록 개선해서 처리시간을 크게 단축했다. 

 

이번 글에서는 직접 코드 리팩토링을 하면서 파이썬 비동기 처리를 하면서 알게된 개념들 위주로 알아보려고 한다.

• async/ await : 비동기 함수 선언 및 호출하기
• asyncio.run() : 동기 함수에서 비동기 함수 호출하기
• asyncio.gather()  vs asyncio.as_completed() : 비동기 처리결과 한번에 반환하기
  • asyncio.gather() : 입력 된 순서를 보장
  • asyncio.as_completed() : 입력된 순서를 보장하지 않고, 완료된 순서대로 반환
• asyncio.get_event_loop() vs asyncio.get_running_loop() : 이벤트 루프 반환하기
  • asyncio.get_event_loop() : 동기함수에서 호출, 없으면 새로 생성. 멀티 스레딩에선 주의요망.
  • asyncio.get_running_loop(): 비동기 함수에서 호출. 멀티 스레딩에서 효율적이고 안정적
  • + loop.run_in_executor(): 멀티 스레딩에서 사용하기 (ThreadPoolExecutor)
• asyncio.Semaphore: 동시 작업 리소스 관리

---

1. async/ await 

코틀린 코투린에서 suspend를 사용해서 비동기 함수를 정의하는 것처럼, 파이썬에서는 async를 사용해서 비동기 함수를 선언한다. 

비동기 함수를 호출할때는 앞에 await을 붙인다. 비동기 함수가 완료되면 다른 처리를 진행할수 있도록 기다린다는 뜻이다. 

그리고 이 비동기 함수를 호출하는 함수 또한 async가 붙은 비동기 함수가 된다.

 

2. asyncio.run() 

당연히 모든 함수가 비동기 함수일수는 없다. 그래서 동기함수에서 비동기 함수를 호출할땐 asyncio.run() 구문을 사용한다. 

asyncio.run() 안에는 실행하고자 하는 비동기 함수를 호출한다. 이때 이벤트 루프를 생성하고, 비동기 작업을 실행한 후 결과를 반환한다.

import asyncio

async def fetch_data(url: str) -> str:
    print(f"Fetching data from {url}...")
    await asyncio.sleep(2)  # 비동기적으로 2초 기다림 (네트워크 요청 시뮬레이션)
    return f"Data from {url}"

async def main():
    url1 = "http://example.com/page1"
    url2 = "http://example.com/page2"
    
    # 두 비동기 작업을 동시에 실행
    result1 = await fetch_data(url1)
    result2 = await fetch_data(url2)
    
    print(result1)
    print(result2)

# 비동기 함수 실행
asyncio.run(main())

 

예를 들어, 이미지 서버에서 request를 여러개 날려서 이미지 response를 받으려고 하는 fetch_data() 함수가 있다.

앞에 async를 붙여 비동기 함수로 선언했다. (위에 코드에서는 네트워크 요청을 동시에 하는 상황을 만들기 위해 2초를 기다리도록 했다)

 

main함수에서는 await을 붙여서 fetch_data() 함수를 호출한다.

요청하고 응답을 받아오는데까지 시간이 걸리기 때문에, 이 작업을 기다리는 동안 다른 작업을 처리할수 있도록 await 해주는 것이다. 

 

비동기 처리를 하는 main함수를 asyncio.run()에 넣어서 실행한다. 

결국 두 작업은 비동기적으로 처리할수 있게 된다.

url1에 대해 요청을 결과를 기다는 동안, url2에 대한 요청을 처리할수 있게 되는 것이다.  

 

3. asyncio.gather()  vs asyncio.as_completed()

파이썬 비동기 작업을 실행할때 자주 사용되는 함수는 asyncio.gather() 와 asyncio.as_completed() 이다.

둘다 비동기작업을 묶어서 병렬로 처리할때, 결과를 한번에 반환하기 위해 사용된다. 

 

그러나 순서를 보장함에 있어서 차이가 있다. 

asyncio.gather()는 순서를 보장한다. 순차적으로 모든 작업이 완료된 후에 순서를 보장한 채 결과를 한번에 반환한다. 

asyncio.as_completed() 순서를 보장하지 않는다. 입력 순서가 아닌 작업이 완료된 순서대로 결과를 한번에 반환한다.

그래서 순서 보장이 필요하다면 asyncio.gather() 를, 빠른 작업부터 처리하고 싶다면 asyncio.as_completed() 를 사용하는것이 좋다.

asyncio.as_completed()를 사용할 때 결과는 완료된 순서대로 반환되므로, 입력 순서를 유지하려면 별도로 정렬이 필요하다.

import asyncio

async def fetch_data(url: str):
    await asyncio.sleep(2)  # 비동기 대기 (네트워크 요청 시뮬레이션)
    return f"Data from {url}"

async def main():
    urls = ["url1", "url2", "url3"]
    
    # 비동기 작업을 병렬로 실행
    results = await asyncio.gather(
        fetch_data(urls[0]),
        fetch_data(urls[1]),
        fetch_data(urls[2])
    )
    print(results)

# 비동기 함수 실행
asyncio.run(main())

 

예를 들어, gather를 사용할 경우 이렇게 입력된 task 순서를 보장해서, 입력된 순서대로 처리하로 결과를 한번에 반환한다. 

import asyncio

async def fetch_data(url: str):
    await asyncio.sleep(2)  # 비동기 대기 (네트워크 요청 시뮬레이션)
    return f"Data from {url}"

async def main():
    urls = ["url1", "url2", "url3"]
    tasks = [fetch_data(url) for url in urls]
    
    for future in asyncio.as_completed(tasks):
        result = await future
        print(result)

# 비동기 함수 실행
asyncio.run(main())

 

그러나, asyncio.as_completed를 사용할 경우 입력된 task 순서는 보장하지 않고, 처리가 빠른 순서대로 반환된다. 

 

4.  asyncio.get_event_loop() vs asyncio.get_running_loop() 

asyncio.run() 을 사용하게 되면 이벤트 루프가 최초로 생성이 된다.

이벤트 루프는 비동기 작업들을 관리하는 중앙 처리 시스템이다.  비동기 작업들을 스케쥴링하고, 작업이 끝다면 다른 작업을 실행할수 

있도록 리소스 관리도 해주면서 비동기 작업들을 병렬로 처리하기 위한 모든 일을 다한다. 

프로그램에서는 단 하나의 이벤트 루프를 관리하며, 멀티 스레딩 환경에서는 각 스레드마다 독립적인 이벤트 루프를 가진다.

 

asyncio.get_event_loop()는 현재 이벤트 루프를 가지고 온다.

동기 코드 내에서만 호출이 가능하며 이벤트 루프가 없으면 새로운 이벤트 루프를 생성한다. 

멀티 스레딩 환경에서 이 함수를 호출하면 다른 스레드에서 이미 생성된 루프를 생성 할수 있어서 주의해야한다. 안전하게 독립적인 이벤트 루프를 사용하려면 메인스레드에서만 호출하던지, asyncio.set_event_loop()로 각 스레드에 별도로 독립적인 루프를 줘야한다. 

 

반면에, asyncio.get_running_loop()는 현재 실행중인 이벤트 루프를 dynamic하게 가지고 온다.

비동기 코드 내에서만 호출이 가능하고, 이벤트 루프가 없으면 RuntimeError이 발생해서 예외처리를 할수 있다.

멀티 스레딩 환경에서 이 함수를 호출하면 각 스레드에서 실행중인 루프를 안전하게 가지고 올 수 있어서 각 스레드마다 독립적인 이벤트 루프를 사용할 수 있다. 특히 멀티 스레딩 환경에서 loop.run_in_executor와 같이 사용하면 별도의 스레드에서 작업을 할수 있다. 

 

 loop.run_in_executor()를 사용해서 비동기 코드내에서 CPU작업이나 블로킹 작업을 별도의 스레드에서 실행할수 있다.

멀티 스레딩에서 병렬작업을 관리하기 위해 ThreadPoolExecutor를 사용할수 있다. 

import asyncio
from concurrent.futures import ThreadPoolExecutor

def blocking_task():
    import time
    time.sleep(2)  # 블로킹 I/O 작업 시뮬레이션
    return "Task completed"

async def main():
    loop = asyncio.get_running_loop()  # 현재 이벤트 루프 가져오기
    with ThreadPoolExecutor(max_workers=4) as executor:
        result = await loop.run_in_executor(executor, blocking_task)  # 별도의 스레드에서 실행
        print(result)

# 비동기 함수 실행
asyncio.run(main())

 

✅ 성능 개선 포인트 ✅

추가적으로 ThreadPoolExecutor는 max_workers를 통해 동시에 실행할수 있는 최대 스레드 수를 조정할수 있다.

기본값은 CPU의 스레드수이고, 보통 max_workers는 CPU 코어수의 2배로 설정하는것이 효율적이다.

 

비동기적으로 처리하는 작업이 네트워크 요청이나 파일을 읽는 IO작업이라면, 스레드수가 많을수록 큰 성능 향상이 예상되나,

CPU를 처리하는 작업이라면, 스레드수가 너무 많을경우 오히려 컨텍스트 스위칭 오버헤드가 발생하여 성능이 저하될수도 있다. 

 

5. asyncio.Semaphore

비동기 처리를 할때 자원관리가 필요하거나 과도한 동시작업으로 리소스가 생각보다 많이 소모가 될수 있다.

이때 asyncio.Semaphore가 필요하다. 동시에 실행 가능한 작업 수를 제한하는 것이다.

예를 들어, 네트워크 요청이나 파일IO 같은 제한된 자원을 사용해야할때 새마포어를 사용해서 과도한 리소스 사용을 방지할수 있다. 

async with semaphore를 사용해서 자원에 대한 접근을 제한해 너무 많은 작업이 동시에 실행되는걸 방지할수 있다.

async def fetch_data(semaphore: asyncio.Semaphore, url: str):
    async with semaphore:
        print(f"Fetching data from {url}...")
        await asyncio.sleep(2)  # 비동기 대기 (예시로 2초 대기)
        return f"Data from {url}"

async def main():
    semaphore = asyncio.Semaphore(2)  # 동시에 실행될 수 있는 작업 수 제한 (2개)
    
    urls = ["url1", "url2", "url3", "url4"]
    tasks = [fetch_data(semaphore, url) for url in urls]
    results = await asyncio.gather(*tasks)
    
    print(results)

asyncio.run(main())

 

✅ 성능 개선 포인트 ✅

asyncio.Semaphore 크기로 비동기적으로 실행하는 작업 수를 조정할수 있다. 

사실 세마포어 없이 비동기 처리를 할수도 있다. 동시 작업수를 아예 제한하지 않는 것이다. 빠를수는 있지만 여러가지 문제가 있을수 있다. 예를 들어 이미지 다운로드 요청을 하는 IO 작업을 비동기 처리를 한다면, 서버로 동시에 너무 많은 요청이 보내져서 서버의 응답속도가 떨어질수도 있고 시스템의 안정성과 성능이 떨어질수 있다. 

 

세마포어 값을 너무 작게 설정하면, 작업의 대기시간이 길어져서 비동기 처리의 장점인 동시성을 제대로 활용하지 못할수 있다. 

기본적으로 세마포어 값을 아예 지정하지 않으면 기본값은 1이다. 동시에 하나의 작업만 실행되서 비동기 처리가 제대로 되지 않는다.

 

반면에 너무 크게 설정하면 완전히 비동기 처리를 할수 있지만, 리소스를 비효율적으로 사용할수 있다. 

세마포어 없이 비동기 처리를 한다면 동시 작업수를 아예 제한하지 않는다. 과도하게 병렬로 실행이 되면서 성능저하, 자원 고갈 등 여러 문제가 발생할수 있기 때문에 적절한 값을 설정해서 동시작업수를 효율적으로 관리하는 것이 중요하다.  

 

---

 

지금까지 파이썬 비동기 처리에 대한 내용을 알아보았다.

실제로 리팩토링하면서 성능이 개선되고 최적화가 되는것도 체감하면서 뿌듯했다. 파이썬 코드를 작성할때 단순히 가독성이 중요한 개발을 넘어서, 이렇게 리소스 관리까지 고려하는게 얼마나 중요한지 다시한번 느꼈다. 역시 지독한 개선의 엔지니어링 작업인가..! 

앞으로도 이렇게 비동기 처리랑 멀티스레딩까지 고려해서 성능을 극대화 할수 있는 방법들을 좀더 알아봐야겠다.  

 

마지막으로 실제로 현업에서 파이썬 코드를 작성하면서 배웠던, 같이 보면 좋은글들까지 남긴다. (이게 벌써 2022년 글이잖아..?!)

2022.09.30-🧐 파이썬 코드를 잘 짜는 법 : 병렬처리 라이브러리 비교분석

2022.08.02-🧐 파이썬 코드를 잘 짜는 법 : 의존성 주입(dependency injector)