asyncio 기본 — async / await / 이벤트 루프

3 동시성 모델 — asyncio 가 I/O 의 현대 디폴트

Python 의 셋 — asyncio (단일 스레드 협력 동시성), threading (여러 스레드, GIL 이 CPU 병렬성 제한), multiprocessing (여러 프로세스, 진짜 병렬성). I/O 바운드 작업 — 네트워크, 파일시스템, 안 그러면 기다릴 모든 곳 — 엔 asyncio 가 현대 디폴트. cwkPippa 가 위에서 아래까지 asyncio.

async def 와 await

async def f(): 가 *coroutine* 정의. 호출해도 본문 안 돌아 — coroutine 객체 반환. await coro 가 다음 일시정지 또는 완료까지 coroutine 실행. await 키워드가 마법 — 이벤트 루프한테 "여기 기다리는 동안 다른 거 돌려도 돼".

asyncio.run — 진입점

asyncio.run(main()) 이 이벤트 루프 생성, coroutine 이 반환할 때까지 실행, 그 다음 루프 종료. 일반 Python 스크립트에서 async 프로그램 시작하는 방법. async 함수 안에선 asyncio.run 호출 X — 그냥 await.

Task — coroutine 동시 실행

await coro 가 *한* coroutine 기다림. 여러 개 동시 실행하려면 각각 asyncio.create_task(coro) 로 Task 에 감싸. await asyncio.gather(*tasks) 가 다 기다림. asyncio.TaskGroup (3.11+) 이 현대 모양 — 적절한 예외 처리 가진 블록 scope task 관리.

원칙: "끝까지 async" 가 진짜 제약. 일반 def 안에서 await 못 함. await 없이 async 함수 호출 못 함 (coroutine 객체 받음, 결과 X). 함수가 async 라고 결정하면 체인 위 모든 호출자도 async.

Code

Hello async world·python

import asyncio

async def hello():
    print("hi")
    await asyncio.sleep(0.5)              # 'sleep' 이 루프에 제어 양보
    print("world")

# 일반 스크립트에서 — 이벤트 루프 시작
asyncio.run(hello())
# hi
# (반초 일시정지)
# world

Task vs sequential — 동시성 이김·python

import asyncio
import time

async def fetch(name, delay):
    print(f"{name}: 시작")
    await asyncio.sleep(delay)
    print(f"{name}: 완료")
    return name

async def sequential():
    start = time.perf_counter()
    a = await fetch("a", 1)
    b = await fetch("b", 1)
    c = await fetch("c", 1)
    print(f"sequential: {time.perf_counter() - start:.2f}s")

async def concurrent():
    start = time.perf_counter()
    results = await asyncio.gather(
        fetch("a", 1),
        fetch("b", 1),
        fetch("c", 1),
    )
    print(f"concurrent: {time.perf_counter() - start:.2f}s")

asyncio.run(sequential())     # ~3 초
asyncio.run(concurrent())     # ~1 초 — 병렬 실행

create_task — 던지고 기억·python

import asyncio

async def background_work():
    await asyncio.sleep(0.5)
    print("백그라운드 완료")

async def main():
    task = asyncio.create_task(background_work())
    print("task 스케줄")
    # 실행 동안 다른 작업
    await asyncio.sleep(0.1)
    print("메인이 다른 거 함")
    await task                              # 백그라운드 완료까지 기다림
    print("다 됨")

asyncio.run(main())

TaskGroup — 현대 구조 동시성 (3.11+)·python

import asyncio

async def fetch(name, delay):
    await asyncio.sleep(delay)
    return f"{name}-result"

async def main():
    async with asyncio.TaskGroup() as tg:
        a = tg.create_task(fetch("a", 0.5))
        b = tg.create_task(fetch("b", 0.3))
        c = tg.create_task(fetch("c", 0.4))
    # 여기서 모든 task 완료, 예외는 ExceptionGroup 로 raise
    print(a.result(), b.result(), c.result())

asyncio.run(main())
# gather + try/except 보다 예외 처리 깔끔

await 까먹기 — 가장 흔한 버그·python

import asyncio

async def add(a, b):
    return a + b

async def main():
    # 잘못 — await 없으면 coroutine, 결과 X
    result = add(2, 3)
    print(type(result))                 # <class 'coroutine'>
    print(result)                       # <coroutine object ...>

    # 맞음
    result = await add(2, 3)
    print(result)                       # 5

asyncio.run(main())

# Python 이 종료 시 안 await 된 coroutine 경고

Exercise

async def fetch(url: str, delay: float) -> str — sleep(delay) await 후 f"got {url}" 반환. 그 다음 async def main() — 다른 delay 로 세 동시 fetch (asyncio.gather 사용). time.perf_counter 로 시간. 그 다음 asyncio.TaskGroup 으로 다시 짜. 둘 다 같은 출력 + 같은 경과 시간 확인.

asyncio 기본 — async / await / 이벤트 루프

3 동시성 모델 — asyncio 가 I/O 의 현대 디폴트

async def 와 await

asyncio.run — 진입점

Task — coroutine 동시 실행

Code

External links

Exercise

Progress

댓글 0