Python 비동기 처리 실전 적용기 - asyncio와 aiohttp

문제 상황

데이터 수집 배치 작업이 점점 느려지고 있었다. 외부 API 500개를 순차 호출하는데 약 10분이 걸렸고, 대부분의 시간이 I/O 대기였다.

import requests

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

asyncio 도입

Python 3.7부터 안정화된 asyncio를 적용하기로 했다. aiohttp를 사용해 비동기 HTTP 요청을 처리했다.

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))

주의사항

동시 요청 수 제한이 필요했다. Semaphore로 동시성을 제어했다.

async def fetch_data(urls, concurrency=50):
    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, return_exceptions=True)

에러 핸들링도 중요했다. 일부 요청 실패가 전체를 멈추지 않도록 return_exceptions=True를 사용했다.

결과

10분 → 2.5분으로 단축됐다. CPU 사용률은 5% 미만으로 유지되며 I/O를 효율적으로 처리했다. 내년에는 다른 배치 작업들도 순차적으로 전환할 예정이다.