상위 목록: 하위 목록: 작성 날짜: 읽는 데 17 분 소요

비동기 처리(Asynchronous Processing)

비동기 처리(Asynchronous Processing)란 어떤 작업이 끝나기를 기다리는 동안 다른 작업을 먼저 진행하는 처리 방식입니다.

네트워크 요청, 파일 입출력, 데이터베이스 조회와 같은 작업은 대부분의 시간을 응답을 기다리는 데 사용합니다.

이러한 작업을 I/O 바운드(I/O Bound) 작업이라 하며, 기다리는 시간 동안 다른 작업을 처리하면 전체 실행 시간을 크게 줄일 수 있습니다.

Python에서는 표준 라이브러리인 asyncio 모듈과 async, await 키워드로 비동기 처리를 구현합니다.

이번 강좌에서는 동기 방식에서 기다리는 시간이 어떻게 낭비되는지 확인한 뒤, 이벤트 루프와 코루틴이 await에서 멈추고 다시 이어지는 과정을 단계별로 살펴봅니다. 이어서 asyncio.gather, 태스크, TaskGroup, 시간 제한, async with, async for, 블로킹 함수 처리, 동시 실행 개수 제한, Python 3.14의 태스크 상태 확인 기능을 다룹니다.


import time


def download(name, delay):
    print(f"{name} 요청")
    time.sleep(delay)
    print(f"{name} 응답 도착")
    return name


start = time.perf_counter()
result = [download("A", 1), download("B", 1), download("C", 1)]
print(result)
print(f"전체 : {time.perf_counter() - start:.2f}s")
print(f"CPU 사용 : {time.process_time():.2f}s")
결과
A 요청
A 응답 도착
B 요청
B 응답 도착
C 요청
C 응답 도착
[‘A’, ‘B’, ‘C’]
전체 : 3.00s
CPU 사용 : 0.06s

비동기 처리가 없을 때의 모습입니다. time.sleep(1)은 서버의 응답을 1초 동안 기다리는 상황을 흉내 낸 것입니다.

download("A", 1)이 응답을 기다리는 1초 동안 프로그램은 아무것도 하지 않고 멈춰 있으며, A가 끝나야 B를 요청합니다. 그래서 세 요청을 처리하는 데 1 + 1 + 1 = 3초가 걸렸습니다.

마지막 줄의 time.process_time()은 프로세스가 실제로 CPU를 사용한 시간입니다. 3초 중 CPU를 사용한 시간은 약 0.06초뿐이고, 나머지는 모두 기다리는 데 사용되었습니다.

세 요청은 서로의 결과가 필요하지 않으므로, A의 응답을 기다리는 동안 B와 C를 미리 요청해 두었다면 세 응답을 동시에 기다려 약 1초 만에 끝낼 수 있습니다. 비동기 처리는 바로 이 기다리는 시간을 겹치게 만드는 방법입니다. 같은 작업을 비동기로 바꾼 결과는 아래의 asyncio.gather 섹션에서 확인합니다.

  • Tip : 실행 시간과 CPU 사용 시간은 실행 환경과 실행할 때마다 달라집니다.


제 31강 쓰레드와 제 46강 프로세스 기반 병렬 처리도 여러 작업을 동시에 처리하는 방법입니다.

세 방식의 차이는 다음과 같습니다.

구분 쓰레드(threading) 프로세스(multiprocessing) 비동기(asyncio)
실행 단위 쓰레드 프로세스 코루틴(태스크)
실행 흐름 여러 쓰레드 여러 프로세스 하나의 쓰레드
작업 전환 운영 체제가 결정 운영 체제가 결정 await 지점에서 코드가 직접 양보
메모리 공유 독립 공유
적합한 작업 I/O 바운드 CPU 바운드 대량의 I/O 바운드
생성 비용 보통 높음 낮음


비동기 처리는 하나의 쓰레드에서 동작하므로 CPU 연산을 병렬로 처리하지는 못합니다.

대신 작업 전환이 await 지점에서만 일어나므로 쓰레드보다 경쟁 상태(Race Condition)를 예측하기 쉽고, 수천 개의 작업을 적은 비용으로 동시에 대기할 수 있습니다.

  • Tip : 계산량이 많은 CPU 바운드(CPU Bound) 작업은 제 46강 프로세스 기반 병렬 처리를 사용합니다.



이벤트 루프와 코루틴

이벤트 루프(Event Loop)는 비동기 처리의 중심이 되는 작업 관리자입니다.

이벤트 루프는 실행할 작업들을 관리하다가, 실행 중인 작업이 기다려야 하는 상황이 되면 대기 중인 다른 작업을 실행합니다.

기다리던 작업이 준비되면 다시 해당 작업을 이어서 실행합니다.


코루틴(Coroutine)은 실행 도중 일시 중지했다가 다시 재개할 수 있는 함수입니다.

제 42강 생성자에서 yield로 함수를 일시 중지하고 send()로 값을 주고받는 코루틴을 다뤘습니다.

asyncio의 코루틴은 이 개념을 비동기 처리에 맞게 발전시킨 것으로, async def로 정의하고 await로 일시 중지합니다.

즉, 코루틴이 await에서 일시 중지하면 이벤트 루프가 다른 코루틴을 실행하며, 이 과정을 반복하여 여러 작업을 동시에 진행하는 것처럼 처리합니다.


class Pause:
    def __await__(self):
        yield "일시 중지"


async def coro():
    print("1단계")
    await Pause()
    print("2단계")
    return "완료"


c = coro()
print(c.send(None))

try:
    c.send(None)
except StopIteration as e:
    print("반환값 :", e.value)
결과
1단계
일시 중지
2단계
반환값 : 완료

코루틴이 await에서 어떻게 멈추는지 확인하기 위해, 이벤트 루프 없이 코루틴을 직접 실행해 보는 예제입니다. 실제 코드에서는 이렇게 사용하지 않으며, 이벤트 루프가 내부에서 하는 일을 손으로 흉내 낸 것입니다.

  1. c.send(None)으로 코루틴을 시작하면 1단계를 출력하고 await Pause()에 도달합니다.
  2. Pause의 __await__가 yield를 실행하는 순간 코루틴은 그 자리에서 멈추고, yield한 값 "일시 중지"가 send()를 호출한 쪽으로 돌아옵니다.
  3. 다시 c.send(None)을 호출하면 멈췄던 위치에서 이어서 2단계를 출력합니다.
  4. 코루틴이 return으로 끝나면 생성자처럼 StopIteration이 발생하고, 반환값은 예외의 value 속성에 담깁니다.

이벤트 루프는 이 send()를 호출하는 쪽입니다. 코루틴이 await asyncio.sleep(1)에서 멈추며 “1초 뒤에 다시 깨워 달라”는 요청을 돌려주면, 이벤트 루프는 그 요청을 기록해 두고 다른 코루틴의 send()를 호출합니다. 1초가 지나면 멈춰 있던 코루틴의 send()를 다시 호출해 이어서 실행합니다.


import asyncio
import time

start = time.perf_counter()


def log(message):
    print(f"{time.perf_counter() - start:.1f}s {message}")


async def worker(name, delay):
    log(f"{name} 시작, {delay}초 대기 요청")
    await asyncio.sleep(delay)
    log(f"{name} 재개")
    return name


async def main():
    log("main 시작")
    t1 = asyncio.create_task(worker("A", 2))
    t2 = asyncio.create_task(worker("B", 1))
    log("태스크 2개 등록")
    await t1
    log("t1 완료")
    await t2
    log("main 종료")


asyncio.run(main())
결과
0.0s main 시작
0.0s 태스크 2개 등록
0.0s A 시작, 2초 대기 요청
0.0s B 시작, 1초 대기 요청
1.0s B 재개
2.0s A 재개
2.0s t1 완료
2.0s main 종료

이번에는 실제 이벤트 루프에서 두 코루틴이 번갈아 실행되는 흐름을 시간과 함께 출력했습니다. create_task()는 아래의 태스크 섹션에서 자세히 다루며, 여기서는 코루틴을 이벤트 루프의 실행 대기열에 등록하는 함수라고만 이해하면 됩니다.

결과를 시간 순서대로 따라가면 이벤트 루프가 하는 일을 볼 수 있습니다.

  1. 0.0초 : asyncio.run()이 이벤트 루프를 만들고 main()을 실행합니다. main()은 태스크 A, B를 대기열에 등록만 하고, 아직 실행하지는 않습니다. 그래서 태스크 2개 등록이 A, B의 시작보다 먼저 출력됩니다.
  2. 0.0초 : main()이 await t1에 도달합니다. t1은 아직 끝나지 않았으므로 main()은 여기서 멈추고 제어권을 이벤트 루프에 돌려줍니다.
  3. 0.0초 : 이벤트 루프는 대기열에서 A를 꺼내 실행합니다. A는 await asyncio.sleep(2)에서 “2초 뒤에 깨워 달라”는 타이머를 등록하고 멈춥니다.
  4. 0.0초 : 이벤트 루프는 다음으로 B를 실행합니다. B도 1초 타이머를 등록하고 멈춥니다.
  5. 0.0초 ~ 1.0초 : 실행할 수 있는 코루틴이 하나도 없습니다. 이벤트 루프는 가장 먼저 끝나는 타이머(1초)까지 운영 체제에 대기를 맡기며, 이 동안 CPU를 거의 사용하지 않습니다.
  6. 1.0초 : B의 타이머가 끝나 B가 깨어나고 B 재개를 출력한 뒤 종료됩니다. t2가 완료 상태가 됩니다.
  7. 2.0초 : A의 타이머가 끝나 A가 깨어나 종료됩니다. t1이 완료되었으므로 await t1에서 멈춰 있던 main()이 다시 실행되어 t1 완료를 출력합니다.
  8. 2.0초 : await t2는 t2가 이미 1초에 끝났으므로 멈추지 않고 바로 결과를 반환하며, main 종료가 출력됩니다.

두 작업을 합치면 3초 분량의 대기지만, 대기 시간이 겹쳤으므로 전체는 약 2초에 끝났습니다.

여기서 중요한 점은 코루틴이 스스로 await에서 멈춰야만 이벤트 루프가 다른 코루틴을 실행할 수 있다는 것입니다. 이벤트 루프는 실행 중인 코루틴을 강제로 중단시키지 못하므로, 코루틴이 await 없이 오래 계산하면 그동안 다른 모든 코루틴이 멈춥니다. 이 문제는 뒤의 블로킹 함수 섹션에서 다룹니다.

  • Tip : 시간은 실행할 때마다 조금씩 달라질 수 있습니다.



async def와 await

import asyncio


async def hello():
    print("Hello")
    await asyncio.sleep(1)
    print("World")
    return "Done"


coro = hello()
print(type(coro))

result = asyncio.run(coro)
print(result)
결과
<class ‘coroutine’>
Hello
World
Done

async def로 정의한 함수를 코루틴 함수라 합니다.

코루틴 함수를 호출하면 함수 내부의 코드가 바로 실행되지 않고, 코루틴 객체가 반환됩니다.

그러므로 hello()를 호출한 시점에는 Hello가 출력되지 않습니다.


asyncio.run(코루틴)은 이벤트 루프를 생성하고 코루틴을 실행한 다음, 실행이 끝나면 이벤트 루프를 닫습니다.

Python 3.7 이상부터 사용할 수 있으며, 프로그램의 진입점에서 한 번만 호출하는 것이 일반적입니다.

await는 코루틴 함수 내부에서만 사용할 수 있으며, await 대상의 실행이 끝날 때까지 현재 코루틴을 일시 중지하고 그 결과를 반환합니다.

asyncio.sleep(초)는 지정한 시간 동안 이벤트 루프를 멈추지 않고 대기하는 코루틴입니다.

  • Tip : await할 수 있는 객체를 어웨이터블(Awaitable)이라 하며, 코루틴, 태스크(Task), 퓨처(Future)가 해당됩니다.

  • Tip : 코루틴 객체를 await하거나 실행하지 않으면 RuntimeWarning: coroutine '...' was never awaited 경고가 발생합니다.

결과를 순서대로 보면, hello()를 호출한 직후에는 <class 'coroutine'>만 출력되고 Hello는 출력되지 않았습니다. asyncio.run(coro)가 코루틴을 실행해야 비로소 Hello가 출력되고, 1초 대기 후 World가 출력됩니다. 마지막 줄은 코루틴이 return한 값을 asyncio.run()이 그대로 반환한 결과입니다.

일반 함수의 time.sleep(1)은 쓰레드 전체를 1초 동안 재우지만, await asyncio.sleep(1)은 이 코루틴만 1초 동안 멈추고 이벤트 루프는 그동안 다른 코루틴을 실행할 수 있습니다. 이 예제에는 다른 코루틴이 없으므로 차이가 보이지 않지만, 다음 섹션부터 그 차이를 확인합니다.


import asyncio


async def get_value():
    await asyncio.sleep(0.1)
    return 10


async def main():
    value = get_value()
    print(type(value).__name__)
    value.close()

    value = await get_value()
    print(value)


asyncio.run(main())
결과
coroutine
10

가장 흔한 실수는 코루틴 함수를 호출하면서 await를 빠뜨리는 것입니다.

await 없이 get_value()를 호출하면 결과값 10이 아니라 코루틴 객체가 반환되며, 함수 본문은 실행되지 않습니다. 예제에서는 경고가 출력되지 않도록 close()로 코루틴 객체를 정리했습니다.

await get_value()로 호출해야 코루틴이 실행되고 반환값 10을 얻을 수 있습니다.


import asyncio


async def inner():
    return "inner"


async def main():
    coro = inner()
    try:
        asyncio.run(coro)
    except RuntimeError as e:
        print("RuntimeError :", e)
    coro.close()
    print(await inner())


asyncio.run(main())
결과
RuntimeError : asyncio.run() cannot be called from a running event loop
inner

asyncio.run()은 새로운 이벤트 루프를 만드는 함수이므로, 이미 이벤트 루프가 실행 중인 코루틴 안에서는 호출할 수 없습니다.

코루틴 안에서 다른 코루틴을 실행하려면 asyncio.run()이 아니라 await를 사용합니다. asyncio.run()은 일반 코드에서 비동기 코드로 들어가는 입구에서 한 번만 사용한다고 기억하면 됩니다.



동시 실행(asyncio.gather)

import asyncio
import time


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


async def sequential():
    await task("A", 1)
    await task("B", 2)
    await task("C", 3)


async def concurrent():
    result = await asyncio.gather(task("A", 1), task("B", 2), task("C", 3))
    print(result)


start = time.perf_counter()
asyncio.run(sequential())
print(f"순차 실행 : {time.perf_counter() - start:.2f}s")

start = time.perf_counter()
asyncio.run(concurrent())
print(f"동시 실행 : {time.perf_counter() - start:.2f}s")
결과
A 시작
A 종료
B 시작
B 종료
C 시작
C 종료
순차 실행 : 6.11s
A 시작
B 시작
C 시작
A 종료
B 종료
C 종료
[‘A’, ‘B’, ‘C’]
동시 실행 : 3.04s

sequential() 함수는 각 코루틴을 await로 하나씩 실행하므로, 앞의 작업이 끝나야 다음 작업이 시작됩니다.

그러므로 전체 실행 시간은 1 + 2 + 3 = 6초입니다.


asyncio.gather(*어웨이터블)은 전달받은 코루틴을 태스크로 등록해 동시에 실행하고, 모두 끝나면 결과를 전달한 순서대로 목록으로 반환합니다.

세 작업이 동시에 대기하므로 전체 실행 시간은 가장 오래 걸리는 작업인 약 3초가 됩니다.

실행 시간은 환경에 따라 조금씩 달라집니다.

동시 실행 결과를 읽어 보면, 세 작업의 시작이 먼저 모두 출력됩니다. A가 await asyncio.sleep(1)에서 멈추면 이벤트 루프가 B를, B가 멈추면 C를 실행하기 때문입니다. 이후 1초, 2초, 3초가 되는 시점에 각각의 타이머가 끝나면서 A, B, C의 순서로 종료가 출력됩니다.

결과 목록 ['A', 'B', 'C']는 끝난 순서가 아니라 전달한 순서입니다. 이 예제는 끝난 순서와 전달한 순서가 같지만, 대기 시간을 3, 2, 1로 바꾸어 C가 먼저 끝나더라도 결과는 ['A', 'B', 'C'] 순서로 반환됩니다.

sequential()에서 await task("A", 1)을 하나씩 실행하면 앞의 코루틴이 끝날 때까지 다음 코루틴은 만들어지지도 않으므로 대기 시간이 겹칠 수 없습니다. 동시에 실행하려면 gather()나 다음 섹션의 태스크처럼 여러 코루틴을 먼저 이벤트 루프에 맡겨 두는 과정이 필요합니다.


import asyncio


async def task(name, delay, fail=False):
    await asyncio.sleep(delay)
    if fail:
        raise ValueError(f"{name} 실패")
    return name


async def main():
    result = await asyncio.gather(
        task("A", 1),
        task("B", 0.5, fail=True),
        task("C", 0.2),
        return_exceptions=True,
    )
    print(result)


asyncio.run(main())
결과
[‘A’, ValueError(‘B 실패’), ‘C’]

asyncio.gather()는 기본적으로 작업 중 하나에서 예외가 발생하면 그 예외를 바로 전달합니다.

return_exceptions=True로 설정하면 예외를 발생시키지 않고, 예외 객체를 결과 목록에 포함해 반환합니다.

결과를 보면 B 자리에 결과값 대신 ValueError('B 실패') 객체가 들어 있고, A와 C는 정상적으로 결과를 반환했습니다. 결과 목록을 반복하면서 isinstance(item, Exception)으로 성공과 실패를 구분해 처리할 수 있습니다.

여러 서버에 요청을 보내고 일부가 실패해도 나머지 결과는 사용해야 하는 경우에 유용합니다.

  • Tip : return_exceptions=False(기본값)일 때 예외가 발생해도 다른 작업은 취소되지 않고 계속 실행됩니다. gather()를 기다리던 쪽에만 첫 번째 예외가 전달됩니다.



태스크(Task)

import asyncio


async def task(name, delay):
    await asyncio.sleep(delay)
    return f"{name} : {delay}s"


async def main():
    t1 = asyncio.create_task(task("A", 2))
    t2 = asyncio.create_task(task("B", 1))
    print(t1.done(), t2.done())

    print(await t2)
    print(t1.done(), t2.done())

    print(await t1)
    print(t1.done(), t2.done())


asyncio.run(main())
결과
False False
B : 1s
False True
A : 2s
True True

태스크(Task)는 코루틴을 이벤트 루프에 등록해 실행을 예약한 객체입니다.

asyncio.create_task(코루틴)으로 태스크를 생성하면, await하지 않아도 이벤트 루프가 실행할 수 있는 시점에 바로 실행을 시작합니다.

done() 메서드로 태스크의 완료 여부를 확인할 수 있으며, 태스크를 await하면 실행 결과를 반환합니다.

t2를 기다리는 1초 동안 t1도 함께 실행되었으므로, t1은 1초만 더 기다리면 완료됩니다.

결과를 한 줄씩 보면, 첫 줄의 False False는 태스크를 만든 직후의 상태입니다. create_task()는 실행을 예약만 하며, main()이 await로 제어권을 넘기기 전까지는 태스크가 한 줄도 실행되지 않기 때문입니다.

await t2에서 main()이 멈추면 두 태스크가 함께 실행되고, 1초 뒤 t2가 끝나 B : 1s가 출력됩니다. 이 시점에 t1은 실행 중이므로 False True입니다. 다시 await t1로 1초를 더 기다리면 두 태스크 모두 완료되어 True True가 됩니다. 전체 소요 시간은 약 2초입니다.

gather()는 여러 코루틴의 결과를 한꺼번에 기다리는 데 적합하고, 태스크는 실행을 먼저 시작해 두고 원하는 시점에 하나씩 결과를 확인하거나, 그 사이에 다른 일을 할 때 적합합니다.

  • Tip : 생성한 태스크는 변수에 저장해 참조를 유지해야 합니다. 이벤트 루프는 태스크를 약한 참조로만 관리하므로 참조가 사라지면 실행 도중 정리될 수 있습니다.



태스크 그룹(TaskGroup)

import asyncio


async def task(name, delay):
    await asyncio.sleep(delay)
    return name


async def main():
    async with asyncio.TaskGroup() as tg:
        t1 = tg.create_task(task("A", 1))
        t2 = tg.create_task(task("B", 2))

    print(t1.result(), t2.result())


asyncio.run(main())
결과
A B

Python 3.11 이상부터 asyncio.TaskGroup으로 여러 태스크를 하나의 그룹으로 관리할 수 있습니다.

async with 블록 안에서 tg.create_task()로 생성한 태스크는, 블록을 벗어날 때 모든 태스크가 완료될 때까지 대기합니다.

블록을 벗어난 후에는 result() 메서드로 각 태스크의 결과를 확인할 수 있습니다.

create_task()로 만든 태스크는 앞의 예제처럼 하나씩 await해야 하고, 실수로 await를 빠뜨리면 태스크가 끝나기 전에 프로그램이 다음 단계로 넘어갈 수 있습니다. TaskGroup은 블록이 끝나는 지점에서 모든 태스크를 자동으로 기다려 주므로, 기다리는 것을 잊을 수 없습니다.

async with 블록이 끝나는 줄이 곧 모든 태스크가 끝나는 지점이므로, 블록 밖의 print()는 2초 뒤에 실행됩니다.


import asyncio


async def task(name, delay, fail=False):
    try:
        await asyncio.sleep(delay)
        if fail:
            raise ValueError(f"{name} 실패")
        print(f"{name} 완료")
    except asyncio.CancelledError:
        print(f"{name} 취소")
        raise


async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(task("A", 0.5))
            tg.create_task(task("B", 1, fail=True))
            tg.create_task(task("C", 3))
    except* ValueError as eg:
        print(repr(eg))
        for e in eg.exceptions:
            print(e)


asyncio.run(main())
결과
A 완료
C 취소
ExceptionGroup(‘unhandled errors in a TaskGroup’, [ValueError(‘B 실패’)])
B 실패

TaskGroup과 asyncio.gather()의 가장 큰 차이는 예외 처리 방식입니다.

그룹 내의 태스크 하나에서 예외가 발생하면, 아직 실행 중인 나머지 태스크(C)를 자동으로 취소합니다.

이후 발생한 예외들을 ExceptionGroup으로 묶어 전달하므로, Python 3.11 이상의 except* 구문으로 예외 종류별로 처리할 수 있습니다.

asyncio.gather()는 예외가 발생해도 나머지 작업을 취소하지 않으므로, 실패한 작업과 상관없이 다른 작업이 계속 실행됩니다.

결과를 시간 순서대로 보면, 0.5초에 A가 정상적으로 끝나 A 완료를 출력합니다. 1초에 B가 ValueError를 발생시키면 TaskGroup은 아직 3초 대기 중인 C를 취소하므로, C의 await 지점에서 CancelledError가 발생해 C 취소가 출력됩니다. 마지막으로 B의 예외가 ExceptionGroup에 담겨 바깥으로 전달됩니다.

C는 원래 3초를 기다려야 했지만, 취소되었으므로 프로그램은 약 1초 만에 끝납니다. 하나가 실패하면 나머지 결과도 의미가 없는 작업이라면, 이처럼 불필요한 대기를 즉시 중단할 수 있습니다.

  • Tip : 일부가 실패해도 나머지 결과를 모두 받아야 한다면 gather(..., return_exceptions=True), 하나라도 실패하면 전체를 중단해야 한다면 TaskGroup을 사용합니다.

  • Tip : 태스크가 취소되면 await 지점에서 asyncio.CancelledError가 발생합니다. 정리 작업을 한 뒤에는 예제처럼 예외를 다시 발생시켜야 취소가 올바르게 전달됩니다.



시간 제한(asyncio.timeout)

import asyncio


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


async def main():
    try:
        async with asyncio.timeout(1):
            print(await slow())
    except TimeoutError:
        print("1초 초과")

    try:
        print(await asyncio.wait_for(slow(), timeout=1))
    except TimeoutError:
        print("wait_for 1초 초과")


asyncio.run(main())
결과
1초 초과
wait_for 1초 초과

Python 3.11 이상부터 asyncio.timeout(초) 비동기 컨텍스트 매니저로 블록 전체에 시간 제한을 설정할 수 있습니다.

제한 시간을 초과하면 블록 안에서 실행 중인 작업을 취소하고 TimeoutError를 발생시킵니다.

코루틴 하나에만 시간 제한을 설정한다면 asyncio.wait_for(코루틴, timeout=초)를 사용할 수도 있습니다.

slow()는 5초가 걸리지만 1초 제한을 걸었으므로, 두 방식 모두 1초 뒤에 slow()가 취소되고 TimeoutError가 발생합니다. 그래서 전체 프로그램은 10초가 아니라 약 2초 만에 끝납니다.

응답하지 않는 서버를 무한히 기다리면 프로그램 전체가 멈춘 것처럼 보이므로, 네트워크 작업에는 항상 시간 제한을 설정하는 것이 좋습니다. 시간 제한은 취소를 통해 구현되므로, 취소된 코루틴 안의 try-finally나 async with의 정리 코드는 정상적으로 실행됩니다.

  • Tip : Python 3.11 이상부터 asyncio.TimeoutError는 내장 예외인 TimeoutError와 같은 클래스이므로 TimeoutError로 처리합니다.



비동기 컨텍스트 매니저(async with)

import asyncio
from contextlib import asynccontextmanager


class Connection:
    def __init__(self, name):
        self.name = name

    async def __aenter__(self):
        await asyncio.sleep(0.1)
        print(f"{self.name} 연결")
        return self

    async def __aexit__(self, exc_type, exc_value, traceback):
        await asyncio.sleep(0.1)
        print(f"{self.name} 연결 해제")
        return False


@asynccontextmanager
async def session(name):
    print(f"{name} 세션 시작")
    try:
        yield name.upper()
    finally:
        print(f"{name} 세션 종료")


async def main():
    async with Connection("DB") as conn:
        print(f"{conn.name} 사용 중")

    async with session("api") as s:
        print(s)


asyncio.run(main())
결과
DB 연결
DB 사용 중
DB 연결 해제
api 세션 시작
API
api 세션 종료

제 48강 컨텍스트 매니저에서 다룬 with 문은 __enter__와 __exit__ 메서드를 사용했습니다.

연결을 열고 닫는 과정에서 네트워크 대기가 필요하다면, async with 문과 비동기 컨텍스트 매니저를 사용합니다.

비동기 컨텍스트 매니저는 __aenter__와 __aexit__ 메서드를 코루틴으로 정의하며, 진입과 종료 시 await를 사용할 수 있습니다.

contextlib.contextmanager와 마찬가지로, contextlib.asynccontextmanager 데코레이터를 사용하면 yield 하나로 비동기 컨텍스트 매니저를 간단하게 작성할 수 있습니다.

일반 with 문의 __enter__에서 네트워크 연결을 기다리면 그 시간 동안 이벤트 루프 전체가 멈춥니다. __aenter__는 코루틴이므로 연결을 기다리는 await asyncio.sleep(0.1) 동안 다른 코루틴이 실행될 수 있습니다.

결과의 순서는 일반 컨텍스트 매니저와 같습니다. DB 연결 → 블록 실행 → DB 연결 해제 순서로 진행되며, session()도 yield 앞의 코드가 진입, finally의 코드가 정리를 담당합니다.

  • Tip : 비동기 컨텍스트 매니저는 async def 함수 안에서 async with로만 사용할 수 있습니다. 일반 with 문에 사용하면 TypeError가 발생합니다.

  • Tip : 앞에서 사용한 asyncio.TaskGroup과 asyncio.timeout도 비동기 컨텍스트 매니저입니다.



비동기 반복(async for)

import asyncio


async def ticker(count, delay):
    for i in range(count):
        await asyncio.sleep(delay)
        yield i


async def main():
    async for value in ticker(3, 0.5):
        print(value)

    result = [value * 10 async for value in ticker(5, 0.1)]
    print(result)


asyncio.run(main())
결과
0
1
2
[0, 10, 20, 30, 40]

async def 함수 안에서 yield를 사용하면 비동기 생성자(Asynchronous Generator)가 됩니다.

비동기 생성자는 값을 하나씩 반환하는 사이에 await로 대기할 수 있으며, async for 문으로 반복합니다.

스트리밍 응답이나 페이지 단위로 데이터를 가져오는 경우처럼, 값마다 대기가 필요한 반복에 활용합니다.

[식 async for 변수 in 비동기 반복자]처럼 리스트 컴프리헨션에서도 사용할 수 있습니다.

결과의 0, 1, 2는 0.5초 간격으로 하나씩 출력됩니다. async for는 다음 값을 요청할 때마다 비동기 생성자를 await하므로, 생성자가 asyncio.sleep()에서 기다리는 동안 이벤트 루프는 다른 코루틴을 실행할 수 있습니다.

일반 for 문으로 비동기 생성자를 반복하면 TypeError가 발생합니다. 일반 반복자는 다음 값을 즉시 돌려줘야 하지만, 비동기 반복자는 다음 값을 기다려야 할 수 있기 때문에 서로 다른 프로토콜을 사용합니다.

  • Tip : 클래스로 구현할 경우 __aiter__와 __anext__ 메서드를 정의하며, 반복이 끝나면 StopAsyncIteration 예외를 발생시킵니다.



블로킹 함수와 asyncio.to_thread

import asyncio
import time


async def blocking(name):
    time.sleep(1)
    return name


async def non_blocking(name):
    await asyncio.to_thread(time.sleep, 1)
    return name


async def main():
    start = time.perf_counter()
    await asyncio.gather(blocking("A"), blocking("B"), blocking("C"))
    print(f"time.sleep : {time.perf_counter() - start:.2f}s")

    start = time.perf_counter()
    await asyncio.gather(non_blocking("A"), non_blocking("B"), non_blocking("C"))
    print(f"to_thread : {time.perf_counter() - start:.2f}s")


asyncio.run(main())
결과
time.sleep : 3.01s
to_thread : 1.18s

asyncio는 하나의 쓰레드에서 이벤트 루프가 실행되므로, 코루틴이 await로 제어권을 양보해야만 다른 작업이 실행됩니다.

time.sleep(), requests.get(), 대용량 파일 읽기처럼 이벤트 루프를 멈추는 함수를 블로킹(Blocking) 함수라 합니다.

코루틴 안에서 블로킹 함수를 호출하면 그 시간 동안 이벤트 루프 전체가 멈추므로, gather()로 묶어도 작업이 순서대로 실행되어 약 3초가 걸립니다.


Python 3.9 이상부터 asyncio.to_thread(함수, *인수)를 사용하면 블로킹 함수를 별도의 쓰레드에서 실행하고, 그 결과를 await로 기다릴 수 있습니다.

이벤트 루프는 멈추지 않으므로 세 작업이 동시에 대기하여 약 1초 만에 완료됩니다. 쓰레드를 준비하는 시간이 더해지므로 정확히 1초가 되지는 않으며, 실행할 때마다 조금씩 달라집니다.

비동기 버전이 없는 라이브러리를 사용해야 할 때 활용합니다.

앞의 이벤트 루프 섹션에서 확인한 것처럼, 이벤트 루프는 코루틴이 await에서 멈춰야만 다른 코루틴으로 넘어갈 수 있습니다. blocking()에는 await가 하나도 없으므로, gather()에 세 개를 넘겨도 이벤트 루프는 A의 time.sleep(1)이 끝날 때까지 B를 시작할 수 없습니다. async def로 정의했다고 해서 내부 코드가 자동으로 비동기가 되는 것은 아닙니다.

non_blocking()은 time.sleep()을 다른 쓰레드에 맡기고 await로 그 결과를 기다립니다. 코루틴이 await에서 멈추므로 이벤트 루프가 다음 코루틴을 실행할 수 있고, 세 쓰레드가 동시에 1초씩 기다립니다.

  • Tip : 비동기 코드에서는 time.sleep() 대신 await asyncio.sleep()을 사용합니다.

  • Tip : asyncio.to_thread는 쓰레드를 사용하므로 CPU 바운드 작업의 속도를 높이지는 못합니다. CPU 바운드 작업은 프로세스 기반 병렬 처리를 사용합니다.



동시 실행 개수 제한(asyncio.Semaphore)

import asyncio
import time

start = time.perf_counter()


async def download(sem, name):
    async with sem:
        print(f"{time.perf_counter() - start:.1f}s {name} 시작")
        await asyncio.sleep(1)
        return name


async def main():
    sem = asyncio.Semaphore(2)
    result = await asyncio.gather(*(download(sem, n) for n in "ABCDE"))
    print(result)
    print(f"전체 : {time.perf_counter() - start:.2f}s")


asyncio.run(main())
결과
0.0s A 시작
0.0s B 시작
1.0s C 시작
1.0s D 시작
2.0s E 시작
[‘A’, ‘B’, ‘C’, ‘D’, ‘E’]
전체 : 3.02s

비동기 처리는 수천 개의 작업을 적은 비용으로 동시에 시작할 수 있지만, 실제로 수천 개의 요청을 한 서버에 동시에 보내면 서버가 요청을 거부하거나 과부하가 걸릴 수 있습니다.

asyncio.Semaphore(개수)는 동시에 블록에 들어갈 수 있는 코루틴의 개수를 제한하는 비동기 컨텍스트 매니저입니다.

결과를 보면 0초에 A, B만 시작하고, 나머지 C, D, E는 async with sem에서 자리가 날 때까지 기다립니다. 1초에 A, B가 끝나 자리가 생기면 C, D가 시작되고, 2초에 E가 시작되어 전체는 약 3초가 걸립니다.

기다리는 코루틴도 await에서 멈춰 있는 상태이므로 이벤트 루프를 막지 않습니다. 실행할 때마다 시간은 조금씩 달라집니다.

  • Tip : 제한이 없다면 5개 모두 0초에 시작해 약 1초 만에 끝납니다. 개수는 상대 서버나 자원의 허용 범위에 맞춰 정합니다.



태스크 상태 확인(Python 3.14)

python -m asyncio ps PID
python -m asyncio pstree PID

Python 3.14 이상부터 실행 중인 다른 Python 프로세스의 asyncio 태스크 상태를 명령줄에서 확인할 수 있습니다.

PID에는 확인할 프로세스의 ID를 입력합니다.

ps는 각 태스크의 태스크 ID, 태스크 이름, 코루틴 호출 스택, 기다리고 있는 태스크(awaiter)를 표 형태로 출력합니다.

pstree는 어떤 태스크가 어떤 태스크를 기다리고 있는지 트리 형태로 출력합니다.

TaskGroup 안의 태스크가 오래 끝나지 않는 경우처럼, 프로그램이 어디에서 대기하고 있는지 확인할 때 유용합니다.

  • Tip : create_task(코루틴, name="이름")으로 태스크에 이름을 지정하면 출력 결과에서 태스크를 쉽게 구분할 수 있습니다.



정리

이름 설명 반환값/특징
async def 코루틴 함수 정의 호출하면 코루틴 객체 반환(본문 미실행)
await 어웨이터블이 끝날 때까지 현재 코루틴 일시 중지 그동안 이벤트 루프가 다른 코루틴 실행
asyncio.run() 이벤트 루프를 만들고 코루틴 실행 프로그램 입구에서 한 번 사용
asyncio.sleep() 이벤트 루프를 막지 않는 대기 time.sleep() 대신 사용
asyncio.gather() 여러 어웨이터블을 동시에 실행 전달한 순서대로 결과 목록 반환
asyncio.create_task() 코루틴 실행 예약 Task 객체, 참조 유지 필요
asyncio.TaskGroup 태스크를 묶어 관리 하나 실패 시 나머지 취소 (Python 3.11 이상)
asyncio.timeout() / wait_for() 시간 제한 초과 시 TimeoutError
async with 비동기 컨텍스트 매니저 __aenter__, __aexit__
async for 비동기 반복 __aiter__, __anext__, 비동기 생성자
asyncio.to_thread() 블로킹 함수를 쓰레드에서 실행 Python 3.9 이상
asyncio.Semaphore 동시 실행 개수 제한 async with로 사용


비동기 처리의 핵심은 기다리는 시간을 겹치는 것입니다. 코루틴은 await에서 스스로 멈추고, 이벤트 루프는 그 사이에 준비된 다른 코루틴을 실행합니다.

그러므로 비동기 코드 안에서는 await 없이 오래 걸리는 코드(블로킹 함수, 긴 계산)를 피하고, 대기가 필요한 곳에는 항상 비동기 버전의 함수나 asyncio.to_thread()를 사용해야 합니다.

댓글 남기기

함께 보면 좋은 글