Async I/O 패턴 — Streaming / 타임아웃 / 취소

Async 반복 — async for / async generator

for x in seq 가 iterable 에 작동하듯, async for x in stream 이 async iterable 에 작동. async generator 는 단지 async def + yield 둘 다 가진 함수 — 각 yield 전에 await 가능. cwkPippa 의 chat 핸들러 작동 방식 — 응답 들어오는 대로 각 청크 yield.

asyncio.timeout — 현대 방법

3.11+ 가 현대 타임아웃 패턴으로 async with asyncio.timeout(seconds): 도입. 3.11 전엔 asyncio.wait_for(coro, timeout). 둘 다 본문이 시간 안에 안 끝나면 TimeoutError raise. timeout context manager 가 다른 async 코드와 합성 더 좋음.

취소 — 뭔가 멈춰야 할 때

task 는 task.cancel() 로 취소 가능. task 안에선 다음 await 지점에 CancelledError raise. 정리 위해 잡을 수 있지만 일반적으로 re-raise 해야. CancelledError 가 일반 예외 X — try/except Exception 안 잡음. 의도적 — 취소가 조용히 삼켜지면 안 됨.

shield — 취소에서 보호

asyncio.shield(coro) 가 안의 coroutine 을 외부에서 전파되는 취소에서 보호. 외부 task 취소되지만 shield 된 안의 거는 계속 실행. "abort 전에 이 중요한 레코드 쓰기 끝내" 에 유용.

War Story: cwkPippa 의 가장 자주 인용되는 함정 — __anext__ 를 asyncio.wait_for 로 절대 감싸지 마. 취소가 async generator 종료, 이후 next 호출이 복구 대신 StopIteration raise. 맞는 모양 — 각 청크 X, 전체 streaming 연산 둘레의 타임아웃 관리.

Code

async generator 위 async for·python

import asyncio

async def number_stream(start, end, delay=0.1):
    for i in range(start, end):
        await asyncio.sleep(delay)
        yield i

async def main():
    async for n in number_stream(0, 5):
        print(n)

asyncio.run(main())
# 0, 1, 2, 3, 4 — 100ms 마다 하나

타임아웃 — 3.11+ 패턴·python

import asyncio

async def slow_op():
    await asyncio.sleep(5)
    return "완료"

async def main():
    try:
        async with asyncio.timeout(1):
            result = await slow_op()
            print(result)
    except TimeoutError:
        print("타임아웃")

asyncio.run(main())

# 3.11 전 방법 (여전히 작동):
async def main_old():
    try:
        result = await asyncio.wait_for(slow_op(), timeout=1)
        print(result)
    except TimeoutError:
        print("타임아웃 (옛 스타일)")

asyncio.run(main_old())

취소 — 잡고 정리 위해 re-raise·python

import asyncio

async def long_task():
    try:
        for i in range(100):
            print("작업 중", i)
            await asyncio.sleep(0.1)
    except asyncio.CancelledError:
        print("취소 받음 — 정리 중")
        # 정리 작업
        raise                               # 취소 전파 위해 re-raise

async def main():
    task = asyncio.create_task(long_task())
    await asyncio.sleep(0.3)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("task 취소 확인")

asyncio.run(main())

shield — 취소에도 뭔가 계속 실행·python

import asyncio

async def critical_save():
    print("저장 중...")
    await asyncio.sleep(0.5)
    print("저장됨")
    return "checkpoint"

async def main():
    task = asyncio.create_task(critical_save())
    try:
        # 외부가 취소돼도 shield 된 task 계속
        result = await asyncio.shield(task)
        print("결과:", result)
    except asyncio.CancelledError:
        print("외부 취소, 안의 task 여전히 실행 중")
        # 보호된 작업이 실제 끝나기 기다림
        result = await task
        print("기다린 후:", result)

asyncio.run(main())

asyncio.Queue — async producer/consumer·python

import asyncio

async def producer(queue, count):
    for i in range(count):
        await asyncio.sleep(0.1)
        await queue.put(f"item-{i}")
        print(f"put item-{i}")
    await queue.put(None)                   # 완료 신호

async def consumer(queue):
    while True:
        item = await queue.get()
        if item is None:
            break
        print(f"got {item}")
        queue.task_done()

async def main():
    queue = asyncio.Queue(maxsize=2)
    await asyncio.gather(producer(queue, 5), consumer(queue))

asyncio.run(main())

Exercise

async def slow_fetch(name, delay) — delay 초 sleep + f"{name}-done" 반환. asyncio.gather 로 둘 동시 실행, 둘 다 asyncio.timeout(1) 에 감싸. delay 0.5 (둘 다 끝남) 와 2 (타임아웃 발동) 로 테스트. 그 다음 TimeoutError 잡고 어느 입력이 일으켰는지 출력. 정리 정확히 일어나는지 확인.

Async I/O 패턴 — Streaming / 타임아웃 / 취소

Async 반복 — async for / async generator

asyncio.timeout — 현대 방법

취소 — 뭔가 멈춰야 할 때

shield — 취소에서 보호

Code

External links

Exercise

Progress

댓글 0