Skip to content

Python의 async/await 문법을 활용한 비동기 프로그래밍 완벽 가이드입니다. asyncio 기본 문법부터 실전 웹 크롤링, Task 제어까지 다루며, 동기 방식 대비 10배 빠른 성능을 얻는 방법을 소개합니다.

비동기 프로그래밍이 필요한 이유

여러분은 파이썬으로 웹 크롤링을 하거나 API를 호출할 때 답답함을 느낀 적 있나요? 한 작업이 끝날 때까지 기다렸다가 다음 작업을 처리하는 동기 방식은 시간이 오래 걸립니다. 특히 네트워크 요청처럼 대기 시간이 긴 작업에서는 CPU가 놀고 있는 시간이 많죠.

비동기 프로그래밍은 이런 대기 시간을 활용해 다른 작업을 동시에 처리합니다. 커피를 주문하고 기다리는 동안 다른 일을 하는 것처럼요. Python 3.5부터 도입된 async/await 문법으로 이제 훨씬 직관적으로 비동기 코드를 작성할 수 있습니다.

핵심 포인트: 비동기는 '병렬 처리'가 아닌 '동시성(Concurrency)' 처리입니다. 하나의 스레드에서 여러 작업을 빠르게 전환하며 실행하는 방식이에요.

asyncio 기본 문법 마스터하기

비동기 프로그래밍의 핵심은 asyncio 라이브러리입니다. 기본 구조를 살펴볼까요?

python import asyncio

async def fetch_data(id): print(f"데이터 {id} 요청 시작") await asyncio.sleep(2) # 네트워크 요청 시뮬레이션 print(f"데이터 {id} 요청 완료") return f"결과 {id}"

async def main(): # 순차 실행 (4초 소요) result1 = await fetch_data(1) result2 = await fetch_data(2)

# 동시 실행 (2초 소요)
results = await asyncio.gather(
    fetch_data(3),
    fetch_data(4)
)

if name == "main": asyncio.run(main())

주요 키워드 이해하기: - async def: 코루틴(Coroutine) 함수를 정의합니다. 비동기로 실행 가능한 함수라는 뜻이에요. - await: 비동기 작업이 완료될 때까지 기다립니다. 이 시간 동안 다른 작업이 실행될 수 있어요. - asyncio.gather(): 여러 코루틴을 동시에 실행하고 모든 결과를 기다립니다. - asyncio.run(): 비동기 프로그램의 진입점입니다.

💡 팁: await는 반드시 async def로 정의된 함수 안에서만 사용할 수 있습니다. 일반 함수에서는 사용할 수 없어요!

실전 활용: 웹 크롤링 성능 개선하기

실제 사례로 여러 웹사이트에서 데이터를 가져오는 코드를 비교해볼게요.

python import aiohttp import asyncio import time

동기 방식 (느림)

def sync_fetch(urls): results = [] for url in urls: # requests.get(url) 등으로 처리 time.sleep(1) # 각 요청마다 1초 results.append(f"Data from {url}") return results # 10개 URL이면 10초 소요

비동기 방식 (빠름)

async def async_fetch(session, url): async with session.get(url) as response: return await response.text()

async def async_fetch_all(urls): async with aiohttp.ClientSession() as session: tasks = [async_fetch(session, url) for url in urls] return await asyncio.gather(*tasks) # 10개 URL도 약 1초

실행

urls = [f"https://example.com/api/{i}" for i in range(10)] results = asyncio.run(async_fetch_all(urls))

동기 방식에서는 10개 요청에 10초가 걸리지만, 비동기 방식에서는 1초면 충분합니다. **약 10배의 성능 향상**이죠!

⚠️ 주의: requests 라이브러리는 동기식이므로 비동기에서는 사용할 수 없습니다. aiohttp, httpx 같은 비동기 지원 라이브러리를 사용하세요.

Task와 Future로 고급 제어하기

더 세밀한 제어가 필요할 때는 Task와 Future 객체를 활용합니다.

python async def advanced_example(): # Task 생성: 백그라운드에서 즉시 실행 시작 task1 = asyncio.create_task(fetch_data(1)) task2 = asyncio.create_task(fetch_data(2))

# 다른 작업 수행 가능
print("Task 실행 중...")

# 필요할 때 결과 수집
result1 = await task1
result2 = await task2

# 타임아웃 설정
try:
    result = await asyncio.wait_for(fetch_data(3), timeout=1.0)
except asyncio.TimeoutError:
    print("시간 초과!")

핵심 차이점: - await fetch_data(): 즉시 실행하고 완료까지 기다립니다. - asyncio.create_task(): 백그라운드에서 실행을 시작하고 Task 객체를 반환합니다. 나중에 await로 결과를 받을 수 있어요.

asyncio.wait_for()를 사용하면 특정 시간 안에 완료되지 않는 작업을 취소할 수 있습니다. 외부 API 호출처럼 응답 시간이 불확실한 경우에 유용해요.

실전 팁과 주의사항

1. CPU 집약적 작업에는 부적합 비동기는 I/O 대기 시간을 활용하는 방식입니다. 복잡한 계산이나 데이터 처리 같은 CPU 집약적 작업에는 효과가 없어요. 이런 경우는 multiprocessing이나 concurrent.futures를 사용하세요.

2. 블로킹 함수 주의 python

잘못된 예

async def bad_example(): time.sleep(1) # 전체 이벤트 루프를 멈춤!

올바른 예

async def good_example(): await asyncio.sleep(1) # 다른 작업 실행 가능

3. 예외 처리 python async def safe_fetch(url): try: async with aiohttp.ClientSession() as session: async with session.get(url) as response: return await response.text() except asyncio.CancelledError: print("작업 취소됨") except Exception as e: print(f"오류 발생: {e}") return None

4. 디버깅 팁 비동기 코드는 실행 흐름이 복잡해서 디버깅이 어렵습니다. asyncio.run(main(), debug=True)로 상세한 로그를 확인하거나, logging 모듈을 적극 활용하세요.

💡 실전 조언: 처음부터 모든 코드를 비동기로 작성하지 마세요. 병목 지점을 찾아 점진적으로 적용하는 것이 효과적입니다. 프로파일링 도구로 성능을 측정하면서 개선하세요!

이 글은 AI가 자동으로 작성했습니다.