asyncio로 동시성 40배 향상 — 그리고 Ops는 우리가 DDoSed라고 생각했다
Source: Dev.to
지난 달, 제품 매니저가 휴게실에서 나를 붙잡고 이렇게 외쳤다. “대시보드가 또 타임아웃이야. 사장이 계속 새로 고침만 하는 거 그만두게 해줄 수 있어?” 그때 우리 메트릭 동기화 스크립트는 실행하는 데 11분이 걸렸다 — 200개의 서드파티 API를 차례대로 호출하고, 로그는 “응답 대기 중”이라는 문구가 줄줄이 쌓였다. 나는 별다른 설명을 하지 않았다. 그냥 책상으로 돌아가 에디터를 열고 생각했다: 이거 비동기로 바꿔야겠어.
일주일 뒤, 리팩터링된 버전이 라이브됐다. 같은 200개의 엔드포인트가 14초 안에 일관되게 완료되었다. 모니터링 알림이 즉시 울렸고, 운영팀은 우리가 DDoS 공격을 당한 줄 알았다. 여기서는 그 리팩터링이 어떻게 작동했는지, asyncio가 빛을 발한 부분과 함정이 된 부분, 그리고 우아하게 문제를 해결하는 방법을 소개한다.
이벤트 루프가 “시간을 훔치는” 방법
asyncio에는 깊은 마법이 없습니다 — 단일 스레드 event loop만 있습니다. 이를 한 가지 일만 정확히 수행하는 초집중 디스패처라고 생각하면 됩니다: 작업 A가 HTTP 요청을 보내고 네트워크를 기다리며 대기 상태가 되면, 디스패처는 이를 일시 중단하고 즉시 작업 B로 넘어가며, A의 바이트가 도착했을 때만 다시 전환합니다. 스레드 전환 오버헤드도 없고, 콜백 지옥도 없습니다 — 모든 로직은 async/await 안에 있습니다.
import asyncio
import aiohttp
import time
# 模拟一次 API 调用
async def fetch_api(session, url: str) -> dict:
async with session.get(url, timeout=aiohttp.ClientTimeout(total=5)) as resp:
return await resp.json()코루틴 함수는 작업을 생성해 이벤트 루프에 넘겨줄 때까지는 단지 설계도에 불과합니다. 여러 코루틴을 한 번에 실행하는 가장 일반적인 방법은 asyncio.gather이며, 한 줄로 모든 코루틴을 동시에 실행합니다. 전체 소요 시간은 이제 모든 요청을 합한 시간이 아니라 가장 오래 걸리는 요청의 시간입니다.
async def main():
urls = [f"https://api.example.com/data/{i}" for i in range(200)]
async with aiohttp.ClientSession() as session:
start = time.time()
# 同时发出 200 个请求，总耗时 ≈ 最慢的一个
tasks = [fetch_api(session, url) for url in urls]
results = await asyncio.gather(*tasks, return_exceptions=True)
elapsed = time.time() - start
print(f"完成 {len(urls)} 个请求，耗时 {elapsed:.2f}s")동기식 버전에서는 200개의 요청이 순차적으로 누적됩니다. 위 코드를 사용하면 모든 연결이 한 번에 대기 상태에 들어가고, 이벤트 루프는 최악의 I/O 시간을 한 번만 지불합니다. 이것이 asyncio가 내 작업 흐름에 가져다준 가장 큰 변화이며, “줄을 서서 기다리는 것이 “한 번에 모두 도착하는 것으로 바뀌었습니다.
Controlling Concurrency and Timeouts So You Don’t Shoot Yourself in the Foot
처음에는 gather를 사용해 모든 작업을 한 번에 실행했는데, 곧 업스트림 게이트웨이의 레이트 제한에 걸려 429 오류가 곳곳에서 발생했습니다. 그 후 asyncio.Semaphore를 도입해 동시 요청 수를 20개로 제한하고, 타임아웃과 재시도 로직을 추가했습니다. 이때 비로소 파이프라인이 실제 운영 환경에 적합해졌습니다.
import asyncio
import aiohttp
from asyncio import Semaphore
CONCURRENCY = 20
MAX_RETRIES = 2
async def fetch_with_limit(sem, session, url):
async with sem: # 超过 20 个协程会在这一行排队
for attempt in range(MAX_RETRIES + 1):
try:
async with session.get(url, timeout=aiohttp.ClientTimeout(total=5)) as resp:
resp.raise_for_status()
return await resp.json()
except Exception as e:
if attempt == MAX_RETRIES:
return {"error": str(e), "url": url}
await asyncio.sleep(2 ** attempt) # 指数退避
async def main_controlled():
urls = [...] # your list of URLs
sem = Semaphore(CONCURRENCY)
async with aiohttp.ClientSession() as session:
tasks = [fetch_with_limit(sem, session, u) for u in urls]
results = await asyncio.gather(*tasks, return_exceptions=True)
return resultsSemaphore는 마치 입구 경비원과 같습니다 — 한 번에 20개의 코루틴만 들어갈 수 있고, 나머지는 await sem에서 대기합니다. 타임아웃, 상태 검사, 지수 백오프 재시도를 추가하면 다운스트림을 과부하 시키지 않으면서 일시적인 오류를 견딜 수 있습니다.
교훈: 조용한 버그를 잡는 3시간
1. await를 잊으면 코루틴이 유령 코드가 된다
네트워크 요청이 실제로 수행되지 않았지만 로그에 “execution successful”라고 표시된 것을 발견했습니다. 반시간 뒤에 fetch_data(url) 대신 await fetch_data(url)를 사용했음을 알게 되었습니다. await가 없는 코루틴은 단순히 제너레이터 객체일 뿐이며, 이벤트 루프는 이를 완전히 무시합니다. Python 3.8+에서는 RuntimeWarning이 발생하지만, 긴 코드 블록에서는 쉽게 놓칠 수 있습니다. 해결 방법: python -W error::RuntimeWarning 옵션으로 경고를 예외로 전환하거나, 명시적으로 asyncio.Task를 생성합니다.
2. 코루틴 내부에 동기 차단 코드를 섞어 사용하기
처음에는 기존 코드를 재사용하면서 async def 안에서 requests.get을 직접 호출했습니다. 그 결과 이벤트 루프가 차단되어 모든 동시성이 직렬 실행으로 떨어졌습니다. 핵심 규칙: async 함수 안에서는 절대 동기 차단 호출을 사용하지 말아야 합니다. 해당 aio 라이브러리(aiohttp, aiofiles 등)로 전환하거나, loop.run_in_executor를 사용해 차단 호출을 스레드 풀로 오프로드하세요.
이 파이프라인을 리팩터링하면서 asyncio의 진정한 힘은 단순히 속도가 아니라, 스레딩의 복잡성 없이 I/O‑바운드 작업을 처리할 수 있다는 점을 깨달았습니다. 하지만 이는 여러분이 규율을 지켜야 함을 의미합니다: 이벤트 루프를 존중하고, 동시성을 제어하며, 언제나 await를 사용하세요.