PHpullh
학습 라이브러리/Python/asyncio 고급 — Queue & Semaphore

PYTHON · 비동기

asyncio 고급 — Queue & Semaphore

비동기 큐와 세마포어로 동시성을 제어하고 프로듀서-컨슈머 패턴을 구현합니다.

비동기고급asyncio.QueueSemaphoreproducerconsumerrate-limit

핵심 설명

비동기 큐와 세마포어로 동시성을 제어하고 프로듀서-컨슈머 패턴을 구현합니다.

Python code

import asyncio

# asyncio.Queue — Producer/Consumer 패턴
async def producer(queue: asyncio.Queue, count: int):
    for i in range(count):
        await asyncio.sleep(0.1)
        await queue.put(f"item-{i}")
        print(f"생산: item-{i}")
    await queue.put(None)  # 종료 신호

async def consumer(queue: asyncio.Queue, name: str):
    while True:
        item = await queue.get()
        if item is None:
            await queue.put(None)  # 다음 컨슈머에게 전달
            break
        await asyncio.sleep(0.2)
        print(f"{name} 소비: {item}")
        queue.task_done()

# Semaphore — 동시 실행 수 제한
async def fetch_with_limit(
    sem: asyncio.Semaphore,
    url: str
) -> str:
    async with sem:   # 동시에 최대 3개만 실행
        await asyncio.sleep(0.1)
        return f"data:{url}"

async def main():
    # Queue 예제
    queue = asyncio.Queue(maxsize=5)
    await asyncio.gather(
        producer(queue, 5),
        consumer(queue, "Worker-1"),
        consumer(queue, "Worker-2"),
    )

    # Semaphore 예제
    sem = asyncio.Semaphore(3)  # 동시 3개 제한
    urls = [f"url-{i}" for i in range(10)]
    results = await asyncio.gather(
        *[fetch_with_limit(sem, url) for url in urls]
    )
    print(f"완료: {len(results)}개")

asyncio.run(main())

학습 팁

asyncio.Semaphore로 외부 API 호출 횟수를 제한하면 rate limit 오류를 방지할 수 있습니다.

주의할 점

queue.task_done()을 호출하지 않으면 queue.join()이 영원히 대기합니다. get() 후 처리 완료 시 반드시 task_done()을 호출하세요.

자주 묻는 질문

asyncio 고급 — Queue & Semaphore란 무엇인가요?

비동기 큐와 세마포어로 동시성을 제어하고 프로듀서-컨슈머 패턴을 구현합니다.

asyncio 고급 — Queue & Semaphore 학습 시 주의할 점은 무엇인가요?

queue.task_done() 을 호출하지 않으면 queue.join() 이 영원히 대기합니다. get() 후 처리 완료 시 반드시 task_done() 을 호출하세요.