Builder.shin

Python 비동기 처리 도입기: asyncio와 aiohttp로 API 호출 최적화

문제 상황

사용자 데이터를 외부 API로 전송하는 배치 작업이 있었다. 1000명의 데이터를 처리하는데 약 10분이 걸렸고, requests 라이브러리로 순차 호출하는 구조였다.

import requests

for user in users:
    response = requests.post(API_URL, json=user.to_dict())
    # 평균 0.6초 소요

asyncio 도입

Python 3.7부터 asyncio가 안정화되면서 비동기 처리를 본격적으로 검토했다. aiohttp를 사용해 동시에 여러 요청을 처리하도록 변경했다.

import asyncio
import aiohttp

async def send_user_data(session, user):
    async with session.post(API_URL, json=user.to_dict()) as response:
        return await response.json()

async def main():
    async with aiohttp.ClientSession() as session:
        tasks = [send_user_data(session, user) for user in users]
        results = await asyncio.gather(*tasks, return_exceptions=True)

asyncio.run(main())

마주친 이슈들

1. 동시 연결 수 제한

처음엔 1000개 요청을 한번에 보내다가 서버에서 429 에러가 발생했다. Semaphore로 동시 연결 수를 20개로 제한했다.

semaphore = asyncio.Semaphore(20)

async def send_with_limit(session, user):
    async with semaphore:
        return await send_user_data(session, user)

2. 예외 처리

일부 요청이 실패해도 전체 작업이 멈추지 않도록 return_exceptions=True 옵션을 사용하고, 결과를 검사해 실패 건을 별도로 처리했다.

3. DB 커넥션 풀

PostgreSQL 조회 부분도 비동기로 전환하려 했으나, 기존 ORM(SQLAlchemy)이 비동기를 지원하지 않아 일단 보류했다. 조회는 동기로 유지하고 API 호출만 비동기 처리했다.

결과

  • 처리 시간: 10분 → 2분
  • 코드 복잡도는 증가했지만 성능 개선 효과가 명확했다
  • 향후 aioredis, asyncpg 등으로 확장 검토 예정

비동기 처리가 만능은 아니지만, I/O 바운드 작업에서는 확실한 효과가 있었다.

Python 비동기 처리 도입기: asyncio와 aiohttp로 API 호출 최적화