Python asyncio로 API 동시 호출 최적화하기

문제 상황

데이터 동기화 배치 작업이 3시간 넘게 걸리는 문제가 있었다. 외부 API를 순차적으로 호출하는 구조였고, 각 요청마다 평균 200ms씩 소요됐다.

# 기존 코드
for item in items:  # 약 50,000개
    response = requests.get(f"{API_URL}/{item.id}")
    process_data(response.json())

해결 방법

asyncio와 aiohttp를 도입해 동시 요청 처리 구조로 변경했다.

import asyncio
import aiohttp

async def fetch_item(session, item_id):
    async with session.get(f"{API_URL}/{item_id}") as response:
        return await response.json()

async def process_batch(items, batch_size=100):
    async with aiohttp.ClientSession() as session:
        for i in range(0, len(items), batch_size):
            batch = items[i:i + batch_size]
            tasks = [fetch_item(session, item.id) for item in batch]
            results = await asyncio.gather(*tasks, return_exceptions=True)
            
            for result in results:
                if isinstance(result, Exception):
                    logger.error(f"Request failed: {result}")
                else:
                    process_data(result)

asyncio.run(process_batch(items))

주의사항

배치 크기를 100으로 제한했다. 처음엔 500으로 설정했다가 API 서버에서 rate limit 에러가 발생했다. asyncio.Semaphore로 동시 연결 수를 제어하는 방법도 고려했지만, 배치 단위 처리가 더 직관적이었다.

에러 핸들링도 중요했다. return_exceptions=True 옵션으로 일부 요청 실패 시에도 전체 배치가 중단되지 않도록 했다.

결과

실행 시간이 3시간에서 40분으로 단축됐다. CPU 사용률은 거의 변화가 없었고, 대부분 I/O 대기 시간이었던 것을 확인했다.