PHpullh
학습 라이브러리/Python/동기→비동기 변환

PYTHON · 비동기

동기→비동기 변환

기존 동기 코드를 비동기로 전환하는 실용적인 패턴과 전략입니다.

비동기중급to_threadrun_in_executor변환동기 비동기

핵심 설명

기존 동기 코드를 비동기로 전환하는 실용적인 패턴과 전략입니다.

Python code

import asyncio
from concurrent.futures import ThreadPoolExecutor
import time

# 동기 함수 (변경 불가한 라이브러리)
def sync_heavy_io(name: str, seconds: float) -> str:
    time.sleep(seconds)  # 블로킹 I/O
    return f"{name} 완료"

def sync_cpu_bound(n: int) -> int:
    return sum(i * i for i in range(n))

# 전략 1: run_in_executor (I/O 바운드)
async def async_io(name: str, seconds: float) -> str:
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(
        None, sync_heavy_io, name, seconds
    )

# 전략 2: ProcessPoolExecutor (CPU 바운드)
async def async_cpu(n: int) -> int:
    loop = asyncio.get_running_loop()
    from concurrent.futures import ProcessPoolExecutor
    with ProcessPoolExecutor() as pool:
        return await loop.run_in_executor(pool, sync_cpu_bound, n)

# 전략 3: asyncio.to_thread (Python 3.9+)
async def main():
    # I/O 바운드 병렬화
    start = time.perf_counter()
    results = await asyncio.gather(
        asyncio.to_thread(sync_heavy_io, "A", 1),
        asyncio.to_thread(sync_heavy_io, "B", 1),
        asyncio.to_thread(sync_heavy_io, "C", 1),
    )
    elapsed = time.perf_counter() - start
    print(f"결과: {results} ({elapsed:.1f}초)")  # ~1초

asyncio.run(main())

학습 팁

asyncio.to_thread()(3.9+)는 run_in_executor의 간편 버전으로, 동기 함수를 스레드에서 실행합니다.

주의할 점

CPU 바운드 작업에 ThreadPoolExecutor를 사용하면 GIL 때문에 성능 향상이 없습니다. ProcessPoolExecutor를 사용하세요.

자주 묻는 질문

동기→비동기 변환란 무엇인가요?

기존 동기 코드를 비동기로 전환하는 실용적인 패턴과 전략입니다.

동기→비동기 변환 학습 시 주의할 점은 무엇인가요?

CPU 바운드 작업에 ThreadPoolExecutor 를 사용하면 GIL 때문에 성능 향상이 없습니다. ProcessPoolExecutor 를 사용하세요.