비동기 흐름의 제한시간과 취소

비동기 제너레이터는 await 사이에서 값을 yield하고 async for가 소비해. Python 3.11의 asyncio.timeout은 블록 전체의 시간 경계를 만들며 이전 버전의 wait_for보다 여러 작업과 조합하기 쉬워.

task.cancel()은 다음 await에서 CancelledError를 일으켜. 정리 뒤에는 보통 다시 올려 취소를 삼키지 말고, 반드시 끝나야 하는 아주 작은 작업만 shield로 보호해.

cwkPippa에서 청크마다 __anext__를 wait_for로 감쌌다가 취소가 비동기 제너레이터를 닫은 사례처럼, 제한시간은 개별 next보다 전체 스트림 수명에 두는 편이 안전해.

Code

비동기 제너레이터와 비동기 반복문·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 마다 하나

블록 전체에 제한시간 두기·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())

취소 뒤 정리하고 다시 올리기·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로 생산자와 소비자 잇기·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())

Code

External links

Exercise

Progress

댓글 0