동시 실행 개수 제어하기 — Semaphore와 Queue

동시 요청 100개를 한꺼번에 쏘면 안 되는 이유

TaskGroup 이나 gather 로 URL 500개를 한 번에 요청하는 코드를 짜면, 문법적으로는 아무 문제가 없습니다. 문제는 상대 서버입니다. 대부분의 API는 초당 요청 수 제한(rate limit)이 있고, 동시 연결 500개를 한꺼번에 받으면 429(Too Many Requests)를 뿌리거나 커넥션을 강제로 끊습니다. asyncio가 "코루틴 수만 개를 동시에 띄울 수 있다"는 사실이, "그렇게 해도 된다"는 뜻은 아닙니다.

asyncio.Semaphore 는 동시에 실행 중인 코루틴 수를 정해진 개수로 제한합니다.

import asyncio
import httpx

sem = asyncio.Semaphore(10)  # 동시에 최대 10개만

async def fetch(client: httpx.AsyncClient, url: str):
    async with sem:
        resp = await client.get(url)
        return resp.status_code

async def main(urls: list[str]):
    async with httpx.AsyncClient() as client:
        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(fetch(client, u)) for u in urls]
    return [t.result() for t in tasks]

500개의 태스크가 동시에 생성되긴 하지만, async with sem 블록에 들어갈 수 있는 건 항상 최대 10개뿐입니다. 나머지는 세마포어가 풀릴 때까지 await 지점에서 대기합니다. 스레드용 threading.Semaphore 와 개념은 같지만, 여기서는 OS 스레드가 아니라 코루틴이 대기열에 줄을 서는 것뿐이라 대기 비용이 훨씬 저렴합니다.

단일 스레드인데도 race condition이 생길 수 있다

"asyncio는 단일 스레드니까 락이 필요 없다"는 말은 반은 맞고 반은 틀립니다. 코드 한 줄 안에서는 중간에 끼어들 수 없는 게 맞지만, await 지점은 다른 코루틴이 끼어들 수 있는 틈입니다.

balance = {"amount": 100}

async def withdraw(amount: int):
    if balance["amount"] >= amount:
        await asyncio.sleep(0)  # 실제로는 DB 호출 같은 게 여기 있다고 가정
        balance["amount"] -= amount
        print(f"{amount}원 출금, 잔액 {balance['amount']}")
    else:
        print("잔액 부족")

async def main():
    await asyncio.gather(withdraw(80), withdraw(80))

asyncio.run(main())

잔액은 100인데 80원 출금 요청이 동시에 둘 들어옵니다. 두 코루틴 모두 if balance["amount"] >= amount 검사를 통과한 뒤(둘 다 100 >= 80이 참) await asyncio.sleep(0) 에서 서로 자리를 바꿉니다. 그 다음 둘 다 80원씩 차감해서 잔액이 -60이 되는, 명백히 잘못된 결과가 나옵니다. 스레드 환경의 race condition과 원인은 다르지만(여긴 진짜 병렬 실행이 아니라 교차 실행 때문) 증상은 똑같습니다.

asyncio.Lock 으로 검사와 차감을 하나의 원자적 구간으로 묶어야 합니다.

lock = asyncio.Lock()

async def withdraw(amount: int):
    async with lock:
        if balance["amount"] >= amount:
            await asyncio.sleep(0)
            balance["amount"] -= amount
            print(f"{amount}원 출금, 잔액 {balance['amount']}")
        else:
            print("잔액 부족")

async with lock 블록 안에 있는 동안에는 다른 코루틴이 같은 락을 얻지 못하고 대기합니다. 검사와 차감 사이에 다른 코루틴이 끼어들 수 없으므로 이번엔 하나는 성공하고 하나는 "잔액 부족"이 됩니다.

생산자-소비자 패턴 — Queue

작업을 만드는 쪽과 처리하는 쪽의 속도가 다를 때는 asyncio.Queue 가 자연스럽습니다.

async def producer(queue: asyncio.Queue):
    for i in range(20):
        await queue.put(f"작업-{i}")
    for _ in range(3):
        await queue.put(None)  # 워커 종료 신호

async def consumer(queue: asyncio.Queue, name: str):
    while True:
        item = await queue.get()
        if item is None:
            break
        print(f"{name} 처리: {item}")

async def main():
    queue = asyncio.Queue(maxsize=5)  # 버퍼 크기 제한
    async with asyncio.TaskGroup() as tg:
        tg.create_task(producer(queue))
        for i in range(3):
            tg.create_task(consumer(queue, f"worker-{i}"))

maxsize=5 를 주면 큐가 가득 찼을 때 producer 의 put 이 자동으로 대기합니다 — 소비자가 못 따라가는 속도로 생산자가 폭주하는 걸 막아줍니다(이른바 배압, backpressure). None 을 종료 신호로 쓰는 건 관용적인 패턴이지 언어 기능은 아닙니다. 워커 개수만큼 None 을 넣어줘야 모든 워커가 정상 종료된다는 걸 놓치기 쉬우니 주의하세요.