Builder.shin

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

문제 상황

운영 중인 Flask 기반 API 서버에서 외부 결제 API와 알림 서비스를 순차적으로 호출하는 엔드포인트의 응답 시간이 평균 3초를 넘어가고 있었다. 각 API 호출이 1초 이상 소요되면서 사용자 경험이 저하되는 상황이었다.

asyncio 도입 검토

동기 방식의 requests 라이브러리를 사용 중이었고, 이를 aiohttp로 전환하기로 결정했다. Python 3.7부터 안정화된 asyncio를 프로덕션에 적용할 시점이라고 판단했다.

구현

import asyncio
import aiohttp
from typing import Dict, List

async def fetch_payment_status(session: aiohttp.ClientSession, order_id: str) -> Dict:
    async with session.get(f'{PAYMENT_API}/orders/{order_id}') as response:
        return await response.json()

async def send_notification(session: aiohttp.ClientSession, user_id: str, message: str) -> Dict:
    async with session.post(f'{NOTIFICATION_API}/send', json={'user_id': user_id, 'message': message}) as response:
        return await response.json()

async def process_order(order_id: str, user_id: str):
    async with aiohttp.ClientSession() as session:
        tasks = [
            fetch_payment_status(session, order_id),
            send_notification(session, user_id, '주문이 처리되었습니다')
        ]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        return results

Flask는 기본적으로 비동기를 지원하지 않아서 asyncio.run()으로 래핑했다. 나중에 FastAPI로 마이그레이션을 고려 중이다.

결과

  • 평균 응답 시간: 3.2초 → 1.4초 (56% 개선)
  • 동시 처리량: 분당 200건 → 450건

주의사항

asyncio.gather()에서 예외 처리가 중요했다. 한 API가 실패해도 다른 작업은 완료되도록 return_exceptions=True 옵션을 사용했다. 또한 connection pool 설정(TCPConnector(limit=100))으로 과도한 연결 생성을 방지했다.

동기 라이브러리와 혼용 시 asyncio.to_thread()를 활용할 수 있지만, 가능하면 전체를 비동기로 통일하는 것이 관리에 유리했다.

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