비동기·스레드·프로세스를 고르는 질문

import asyncio
import aiohttp                         # async HTTP 클라이언트

async def fetch(session, url):
    async with session.get(url) as resp:
        return await resp.text()

async def fetch_all(urls):
    async with aiohttp.ClientSession() as session:
        return await asyncio.gather(*(fetch(session, u) for u in urls))

# (데모만 — asyncio.run(fetch_all(urls)) 로 실행)
# 수백 동시 fetch, 단일 스레드, 한 이벤트 루프

from concurrent.futures import ProcessPoolExecutor

def heavy_compute(n):
    total = 0
    for i in range(n):
        total += i * i
    return total

if __name__ == '__main__':
    with ProcessPoolExecutor() as pool:
        results = list(pool.map(heavy_compute, [10**7] * 4))
        print(results)
# 4 코어 실제 병렬 사용, GIL 제약 없음

import asyncio

def heavy_compute(n):                  # sync, CPU 바운드
    total = 0
    for i in range(n):
        total += i * i
    return total

async def main():
    # 이벤트 루프 안 막게 스레드 (또는 프로세스) 로 offload
    result = await asyncio.to_thread(heavy_compute, 10_000_000)
    print(result)

asyncio.run(main())

# 프로세스 레벨엔 — loop.run_in_executor(ProcessPoolExecutor(), fn, *args)

# threading 이 이기는 특정 케이스:
# - sync 라이브러리 사용해야 (async 버전 없음)
# - 지금 async 로 리팩터 안 하는 중
# - 작업이 I/O 바운드
#
# 예 — sync 만인 legacy DB 드라이버.
# asyncio.to_thread 또는 ThreadPoolExecutor 로 감싸.
#
# threading을 무턱대고 기본값으로 삼지 마 — 사람들 가정보다 적은 경우에 답.
# asyncio 가 거의 모든 현대 I/O 라이브러리 지원,
# threading 은 갭 메우기.