Python asyncio로 API 병렬 호출 성능 개선하기

문제 상황

매일 새벽 돌아가는 데이터 수집 배치 작업이 있었다. 외부 API를 호출해 약 500개 항목의 데이터를 가져오는데, 순차 처리 방식이라 10분 이상 소요됐다.

# 기존 코드
def fetch_all_data(ids):
    results = []
    for id in ids:
        response = requests.get(f'https://api.example.com/items/{id}')
        results.append(response.json())
    return results

각 API 호출이 1~2초 걸리는데 순차 처리하니 당연히 느렸다.

해결 방법

aiohttp와 asyncio를 사용해 병렬 처리로 변경했다.

import asyncio
import aiohttp

async def fetch_item(session, id):
    async with session.get(f'https://api.example.com/items/{id}') as response:
        return await response.json()

async def fetch_all_data(ids):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_item(session, id) for id in ids]
        return await asyncio.gather(*tasks)

# 실행
results = asyncio.run(fetch_all_data(ids))

주의사항

처음엔 동시 요청 수 제한 없이 실행했다가 API 서버에서 429 에러가 발생했다. Semaphore로 동시 요청 수를 20개로 제한하니 안정적으로 동작했다.

async def fetch_all_data(ids, limit=20):
    semaphore = asyncio.Semaphore(limit)
    
    async def fetch_with_limit(session, id):
        async with semaphore:
            return await fetch_item(session, id)
    
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_with_limit(session, id) for id in ids]
        return await asyncio.gather(*tasks)

결과

• 실행 시간: 10분 → 1분 30초
• 동시 요청 수 제한으로 API 서버 부하 관리
• 기존 동기 코드 대비 약간의 복잡도 증가는 있지만 성능 개선이 확실했다

I/O 바운드 작업에서 asyncio는 확실히 효과적이었다.