Builder.shin

Python asyncio로 API 병렬 호출 최적화하기

문제 상황

재택근무 전환 이후 데이터 동기화 배치 작업의 실행 시간이 문제가 되었다. 약 500개의 사용자 데이터를 외부 API로 조회하는데, 순차 처리로 인해 25분 이상 소요되고 있었다.

기존 코드는 단순한 for문 반복이었다.

def sync_user_data(user_ids):
    results = []
    for user_id in user_ids:
        response = requests.get(f'{API_URL}/users/{user_id}')
        results.append(response.json())
    return results

asyncio와 aiohttp 적용

Python 3.7 이상에서 안정화된 asyncio를 사용하기로 했다. aiohttp로 비동기 HTTP 요청을 처리했다.

import asyncio
import aiohttp

async def fetch_user(session, user_id):
    async with session.get(f'{API_URL}/users/{user_id}') as response:
        return await response.json()

async def sync_user_data(user_ids):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_user(session, user_id) for user_id in user_ids]
        results = await asyncio.gather(*tasks, return_exceptions=True)
    return results

# 실행
results = asyncio.run(sync_user_data(user_ids))

개선 사항

  • 실행 시간: 25분 → 5분
  • 동시 요청 수 제한이 필요해 semaphore 추가
  • return_exceptions=True로 일부 실패해도 전체 중단 방지
async def sync_user_data(user_ids, concurrency=50):
    semaphore = asyncio.Semaphore(concurrency)
    
    async def fetch_with_limit(session, user_id):
        async with semaphore:
            return await fetch_user(session, user_id)
    
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_with_limit(session, uid) for uid in user_ids]
        results = await asyncio.gather(*tasks, return_exceptions=True)
    return results

asyncio는 I/O 바운드 작업에서 효과가 확실했다. CPU 집약적인 작업엔 multiprocessing이 맞겠지만, API 호출이 대부분인 배치 작업에선 이 방식이 충분했다.

Python asyncio로 API 병렬 호출 최적화하기