Python 비동기 처리와 asyncio 실전 적용기

배경

레거시 Python API 서버가 외부 API 호출이 많아지면서 응답 속도가 느려졌다. 순차적으로 실행되던 I/O 작업들을 병렬화하기 위해 asyncio를 도입하기로 결정했다.

기존 코드의 문제

def fetch_user_data(user_id):
    profile = requests.get(f'/api/profile/{user_id}')
    orders = requests.get(f'/api/orders/{user_id}')
    reviews = requests.get(f'/api/reviews/{user_id}')
    return combine_data(profile, orders, reviews)

3개의 API 호출이 순차적으로 실행되어 총 900ms가 소요됐다.

asyncio 적용

import asyncio
import aiohttp

async def fetch_user_data(user_id):
    async with aiohttp.ClientSession() as session:
        tasks = [
            session.get(f'/api/profile/{user_id}'),
            session.get(f'/api/orders/{user_id}'),
            session.get(f'/api/reviews/{user_id}')
        ]
        results = await asyncio.gather(*tasks)
        return combine_data(*results)

병렬 처리로 응답 시간이 300ms로 단축됐다.

마주친 문제들

1. 동기 라이브러리 혼재

requests는 동기 라이브러리라 async 함수 내에서 사용하면 의미가 없었다. aiohttp로 전면 교체했다.

2. DB 커넥션 이슈

SQLAlchemy의 기본 엔진은 동기식이었다. databases 라이브러리를 추가해 비동기 쿼리를 구현했다.

from databases import Database

database = Database('postgresql://...')

async def get_user(user_id):
    query = "SELECT * FROM users WHERE id = :user_id"
    return await database.fetch_one(query, values={"user_id": user_id})

3. 에러 핸들링

asyncio.gather()는 하나의 태스크가 실패하면 전체가 실패한다. return_exceptions=True 옵션으로 개별 처리가 가능했다.

결과

평균 응답 시간 67% 개선, 동시 처리 가능 요청 수 3배 증가. 다만 코드 복잡도가 올라가 팀원들의 학습 곡선이 필요했다. 모든 곳에 asyncio를 적용할 필요는 없고, I/O 병목이 명확한 부분에 선택적으로 적용하는 게 합리적이었다.