Python asyncio로 API 병렬 처리 성능 개선

문제 상황

데이터 동기화 배치 작업이 외부 API를 약 2000번 호출하는데, 순차 처리로 인해 30분 이상 소요되고 있었다. 각 API 호출은 I/O 대기 시간이 대부분이라 병렬 처리로 개선 여지가 충분했다.

asyncio 적용

기존 requests 대신 aiohttp를 사용해 비동기 처리로 전환했다.

import asyncio
import aiohttp
from typing import List

async def fetch_data(session: aiohttp.ClientSession, item_id: str):
    url = f"https://api.example.com/items/{item_id}"
    try:
        async with session.get(url, timeout=10) as response:
            return await response.json()
    except asyncio.TimeoutError:
        print(f"Timeout: {item_id}")
        return None
    except Exception as e:
        print(f"Error {item_id}: {e}")
        return None

async def process_batch(item_ids: List[str]):
    connector = aiohttp.TCPConnector(limit=50)
    async with aiohttp.ClientSession(connector=connector) as session:
        tasks = [fetch_data(session, item_id) for item_id in item_ids]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        return [r for r in results if r is not None]

if __name__ == "__main__":
    item_ids = [...]  # 2000개 항목
    results = asyncio.run(process_batch(item_ids))

동시성 제어

TCPConnector의 limit으로 동시 연결 수를 50으로 제한했다. 처음엔 제한 없이 실행했다가 서버 측에서 429 에러가 발생했다. Semaphore를 추가해 더 세밀한 제어도 가능하지만, 현재는 connector limit만으로 충분했다.

결과

• 실행 시간: 30분 → 4분 30초
• 에러율: timeout 설정으로 일부 느린 응답 스킵
• 재시도 로직은 별도로 구현해 실패 건만 처리

처음 asyncio를 적용할 땐 이벤트 루프 개념이 낯설었지만, I/O 바운드 작업에서는 확실히 효과적이었다. 다만 디버깅이 동기 코드보다 까다로워 로깅을 꼼꼼히 남기는 게 중요했다.