Builder.shin

Python asyncio와 aiohttp로 API 병렬 호출 최적화

문제 상황

데이터 수집 배치 작업이 있었다. 외부 API 100여 개를 순차적으로 호출하는데 평균 25분이 소요됐다. 각 API 응답 시간은 2~3초 정도였지만, 순차 처리라 전체 시간이 길었다.

기존 코드는 requests 라이브러리로 단순하게 구현되어 있었다.

import requests

def fetch_data(urls):
    results = []
    for url in urls:
        response = requests.get(url)
        results.append(response.json())
    return results

asyncio 전환

aiohttp를 사용해 비동기 처리로 전환했다. Python 3.9 환경이라 async/await 문법을 안정적으로 사용할 수 있었다.

import asyncio
import aiohttp

async def fetch_one(session, url):
    async with session.get(url) as response:
        return await response.json()

async def fetch_data(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_one(session, url) for url in urls]
        return await asyncio.gather(*tasks)

# 실행
results = asyncio.run(fetch_data(urls))

에러 핸들링

초기 구현에서 일부 API가 타임아웃되면 전체 작업이 실패하는 문제가 있었다. return_exceptions=True 옵션으로 해결했다.

async def fetch_one(session, url):
    try:
        async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as response:
            return await response.json()
    except Exception as e:
        return {"error": str(e), "url": url}

async def fetch_data(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_one(session, url) for url in urls]
        return await asyncio.gather(*tasks, return_exceptions=True)

결과

  • 실행 시간: 25분 → 5분
  • 에러율: 동일 (타임아웃 발생 건수 변화 없음)
  • 메모리 사용량: 큰 차이 없음

동시 연결 수가 많아 일부 API에서 rate limit 에러가 발생했다. asyncio.Semaphore로 동시 요청 수를 20개로 제한하니 안정화됐다.

async def fetch_data(urls, concurrency=20):
    semaphore = asyncio.Semaphore(concurrency)
    
    async def fetch_with_limit(session, url):
        async with semaphore:
            return await fetch_one(session, url)
    
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_with_limit(session, url) for url in urls]
        return await asyncio.gather(*tasks)

I/O bound 작업에서는 asyncio가 확실히 효과적이었다.

Python asyncio와 aiohttp로 API 병렬 호출 최적화