asyncio와 httpx

src/content/documents/python/python-asyncio-httpx-concurrent-collection.json

24편에서 I/O-bound 작업에는 threading 이 효과적이라고 확인했다. 그런데 API를 100개 부른다고 스레드를 100개 만드는 것도 부담이다. 스레드 하나마다 OS가 스택 메모리를 잡고 전환 비용을 문다. asyncio 는 스레드를 늘리지 않고, 하나의 스레드 안에서 대기 시간이 겹치도록 여러 요청을 번갈아 진행시킨다.

한 줄 정의

asyncio 와 httpx 의 조합은 하나의 스레드에서 이벤트 루프가 여러 네트워크 요청의 대기 시간을 겹쳐 처리하는 비동기 I/O 방식이다. async def 로 정의한 코루틴이 await 를 만날 때마다 실행을 이벤트 루프에 양보하고, 그 사이 다른 코루틴이 진행된다.

threading과 무엇이 다른가

구분

threading

asyncio

실행 단위

OS 스레드 (요청마다 스레드 필요)

코루틴 (하나의 스레드 안에서 전환)

전환 비용

OS가 스레드를 스케줄링

이벤트 루프가 코드 안에서 직접 전환

코드 형태

일반 함수 그대로 사용

async/await로 명시해야 함

적합 규모

동시 요청 수십 개

동시 요청 수백~수천 개까지 가볍게

스레드는 몇 개만 써도 만들고 전환하는 비용이 눈에 띈다. 코루틴은 스레드보다 훨씬 가벼워서, 공공 API 수십~수백 개를 한 번에 호출하는 수집기에는 asyncio 쪽이 더 알맞다.

async와 await

  • async def 로 정의한 함수는 호출해도 즉시 실행되지 않고 코루틴(coroutine) 객체 를 반환한다.

  • await 는 그 코루틴이 끝날 때까지 기다리되, 기다리는 동안 이벤트 루프가 다른 코루틴을 진행하도록 실행을 양보한다.

  • asyncio.run() 이 이벤트 루프를 만들고 최상위 코루틴을 실행한 뒤 루프를 정리한다. 보통 스크립트에서 한 번만 호출한다.

import asyncio

async def fetch_one():
    await asyncio.sleep(1)  # 실제로는 네트워크 응답을 기다리는 자리
    return "결과"

asyncio.run(fetch_one())

httpx.AsyncClient로 요청 하나 보내기

httpx 는 동기 Client 와 비동기 AsyncClient 를 같은 API로 제공한다. async with 로 클라이언트를 열어두고 그 안에서 여러 요청을 보낸다 — 요청마다 클라이언트를 새로 만들면 연결을 재사용하지 못해 오히려 느려진다.

import asyncio
import httpx

async def fetch_weather():
    async with httpx.AsyncClient() as client:
        response = await client.get(
            "https://api.open-meteo.com/v1/forecast",
            params={"latitude": 37.5665, "longitude": 126.9780,
                    "hourly": "temperature_2m", "forecast_days": 1,
                    "timezone": "Asia/Seoul"},
            timeout=10,
        )
        return response.json()["hourly"]["temperature_2m"][:3]

print(asyncio.run(fetch_weather()))
[26.6, 26.3, 26.1]

asyncio.gather로 여러 요청을 동시에

서로 관련 없는 API 여러 개를 부를 때는 코루틴을 만들어 두고 asyncio.gather 에 한꺼번에 넘긴다. 하나씩 await 하지 않고 모든 태스크를 동시에 시작시킨 뒤 다 끝나기를 기다린다.

import asyncio
import time

import httpx

URLS = [
    "https://api.open-meteo.com/v1/forecast?latitude=37.5665&longitude=126.9780"
    "&hourly=temperature_2m&forecast_days=1&timezone=Asia/Seoul",
    "https://countries.dev/alpha/KOR",
    "http://ip-api.com/json/8.8.8.8",
    "https://api.open-meteo.com/v1/does-not-exist",  # 일부러 실패를 넣는다
]


async def fetch(client: httpx.AsyncClient, url: str) -> dict:
    try:
        response = await client.get(url, timeout=10)
        response.raise_for_status()
        return {"url": url, "ok": True, "data": response.json()}
    except httpx.HTTPStatusError as e:
        return {"url": url, "ok": False, "error": f"HTTP {e.response.status_code}"}
    except httpx.HTTPError as e:
        return {"url": url, "ok": False, "error": str(e)}


async def fetch_all(urls: list[str]) -> list[dict]:
    async with httpx.AsyncClient() as client:
        tasks = [fetch(client, url) for url in urls]
        return await asyncio.gather(*tasks)


async def main():
    start = time.perf_counter()
    results = await fetch_all(URLS)
    elapsed = time.perf_counter() - start

    ok = [r for r in results if r["ok"]]
    failed = [r for r in results if not r["ok"]]
    print(f"{len(URLS)}개 요청, {elapsed:.2f}초 — 성공 {len(ok)} / 실패 {len(failed)}")
    for r in failed:
        print(f"  실패: {r['url']} -> {r['error']}")


asyncio.run(main())
4개 요청, 1.13초 — 성공 3 / 실패 1
  실패: https://api.open-meteo.com/v1/does-not-exist -> HTTP 404

부분 실패를 다루는 두 가지 방법

여러 요청 중 하나가 실패했다고 전체 수집이 멈추면 곤란하다. fetch 함수 안에서 예외를 잡아 항상 dict를 반환 하게 만들면, 호출부는 예외 처리를 신경 쓰지 않고 ok 값만 확인하면 된다 — 위 코드가 이 방식이다.

함수 내부에서 예외를 잡지 않고 그대로 전파하고 싶다면 asyncio.gather 에 return_exceptions=True 를 준다. 이때는 실패한 자리에 예외 객체가 그대로 들어오므로, 결과를 순회하며 isinstance(r, Exception) 로 성공·실패를 구분해야 한다.

results = await asyncio.gather(
    *[fetch_raises(client, u) for u in urls],
    return_exceptions=True,
)
for r in results:
    if isinstance(r, Exception):
        print("실패:", r)
    else:
        print("성공:", r)

return_exceptions=True 없이 예외를 던지는 코루틴을 gather 에 넘기면, 그 순간 예외가 즉시 위로 전파되고 다른 태스크의 완료 여부와 무관하게 gather 호출 자체가 실패한다. 부분 실패를 허용하려면 둘 중 하나를 반드시 선택해야 한다.

실측: 순차 요청 vs 동시 요청

같은 API 3개를 동기 httpx.Client 로 순서대로 부른 경우와, 비동기로 동시에 부른 경우를 비교한다.

# 순차 (httpx.Client)
with httpx.Client() as client:
    for url in URLS:
        client.get(url, timeout=10)

# 동시 (httpx.AsyncClient + gather)
async with httpx.AsyncClient() as client:
    await asyncio.gather(*[client.get(u, timeout=10) for u in URLS])
순차 요청 3개: 2.13초
동시 요청 3개: 0.98초

요청이 3개뿐이라 차이가 크지 않아 보이지만, 순차 방식은 요청 수에 비례해서 시간이 늘어나는 반면 동시 방식은 가장 느린 응답 하나에 수렴한다. API를 100개 부른다면 순차는 각 응답 시간의 합, 동시 요청은 그중 가장 느린 응답 하나에 가까운 시간으로 끝난다는 뜻이다.

장점과 한계

구분

내용

장점

스레드보다 가벼워 동시 요청 수백 개도 부담이 적고, 데이터 수집처럼 대기 시간이 대부분인 작업에 특히 잘 맞는다.

한계

CPU-bound 연산에는 도움이 안 된다(24편 참고). 코드 전체가 async 함수 체인으로 이어져야 해서, 기존 동기 코드와 섞기 까다롭고 디버깅도 동기 코드보다 한 단계 더 신경 써야 한다.

자주 발생하는 문제

await를 빼먹는다. 코루틴은 호출만 해서는 실행되지 않는다. await 없이 호출하면 코루틴 객체만 만들어지고 실제 실행은 되지 않은 채 경고가 남는다.

async def main():
    fetch()  # await를 빠뜨렸다
    print("main 끝")

asyncio.run(main())
RuntimeWarning: coroutine 'fetch' was never awaited
main 끝

async 함수 안에서 동기 sleep을 쓴다. time.sleep 은 이벤트 루프에 실행을 양보하지 않고 스레드 자체를 멈춘다. 동시에 실행되던 다른 코루틴도 함께 멈춘다.

async def bad_task():
    time.sleep(1)   # 이벤트 루프 전체를 막는다

async def good_task():
    await asyncio.sleep(1)  # 제어권을 돌려준다

# bad_task 2개 동시 실행: 2.01초 (사실상 순차)
# good_task 2개 동시 실행: 1.00초 (동시에 대기)

네트워크 호출도 마찬가지다. 동기 requests 라이브러리를 async def 안에서 그대로 쓰면 같은 문제가 생긴다. 비동기 코드 안에서는 httpx.AsyncClient 처럼 await 를 지원하는 비동기 라이브러리를 써야 한다.

요청마다 클라이언트를 새로 만든다. httpx.AsyncClient() 를 요청마다 새로 열면 TCP 연결을 매번 새로 맺어 오히려 느려진다. 여러 요청을 묶어 처리할 때는 클라이언트 하나를 열어두고 그 안에서 재사용한다.

관련 문서

I/O-bound와 CPU-bound를 나누는 기준, threading 과 multiprocessing 을 선택하는 기준은 동시성·병렬성·GIL 한 번에 이해하기 에서 다뤘다. 수집한 데이터를 검증하는 방법은 Pydantic으로 외부 데이터를 검증하기 를 잇는다.

참고 자료

댓글 0

댓글을 불러오는 중…