Python 비동기 프로그래밍, 이제는 제대로 이해하고 써보자

Python의 비동기 프로그래밍 기초부터 실전 활용까지 완벽 정리. async/await 문법, 비동기 웹 크롤링 구현, 실무 함정과 해결법을 예제 코드와 함께 소개합니다.

비동기 프로그래밍이 왜 필요할까?¶

웹 크롤링을 하거나 API 요청을 여러 번 보낼 때, 한 번의 요청이 끝날 때까지 기다리는 건 정말 비효율적이죠. 마치 커피숍에서 한 명의 주문이 완전히 끝날 때까지 다음 손님을 받지 않는 것과 같아요.

비동기 프로그래밍은 이런 대기 시간을 활용해 다른 작업을 동시에 처리하는 방식입니다. Python에서는 asyncio 라이브러리를 통해 이를 구현할 수 있어요. I/O 작업(네트워크 요청, 파일 읽기 등)이 많은 프로그램에서 특히 빛을 발하죠.

핵심 포인트: 비동기는 멀티스레딩과 다릅니다. 하나의 스레드에서 작업을 전환하며 효율적으로 처리하는 방식이에요.

async/await 문법의 기초¶

비동기 프로그래밍의 핵심은 async와 await 키워드입니다. 기본 구조를 먼저 살펴볼까요?

python import asyncio

async def fetch_data(id): print(f"작업 {id} 시작") await asyncio.sleep(2) # 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)
)

asyncio.run(main())

async def로 정의된 함수는 코루틴(coroutine)이 되며, 반드시 await를 통해 호출해야 합니다. await 키워드는 "이 작업이 끝날 때까지 기다리되, 그 동안 다른 작업을 해도 돼"라고 말하는 것과 같아요.

💡 TIP: asyncio.gather()를 사용하면 여러 코루틴을 동시에 실행하고 모든 결과를 한 번에 받을 수 있습니다.

실전 예제: 비동기 웹 크롤링¶

실제로 가장 많이 활용되는 사례인 웹 크롤링을 비동기로 구현해볼게요. aiohttp 라이브러리를 사용합니다.

python import aiohttp import asyncio import time

async def fetch_url(session, url): async with session.get(url) as response: data = await response.text() return len(data)

async def main(): urls = [ 'https://example.com', 'https://python.org', 'https://github.com' ] * 10 # 30개 URL

start_time = time.time()

async with aiohttp.ClientSession() as session:
    tasks = [fetch_url(session, url) for url in urls]
    results = await asyncio.gather(*tasks)

elapsed = time.time() - start_time
print(f"총 {len(results)}개 페이지 크롤링 완료")
print(f"소요 시간: {elapsed:.2f}초")

asyncio.run(main())

동기 방식으로 30개 URL을 처리하면 수십 초가 걸리지만, 비동기로 처리하면 불과 몇 초 안에 완료됩니다. 실제 프로젝트에서 이 차이는 엄청나죠.

주의사항: aiohttp의 세션은 반드시 async with로 관리해야 리소스가 제대로 해제됩니다.

비동기 프로그래밍의 함정과 해결법¶

비동기 코드를 작성하다 보면 자주 마주치는 문제들이 있어요.

1. 동기 함수 호출 문제

비동기 함수 안에서 time.sleep() 같은 동기 함수를 호출하면 전체 이벤트 루프가 멈춥니다. 반드시 비동기 버전(asyncio.sleep())을 사용하세요.

2. 예외 처리

python async def safe_fetch(session, url): try: async with session.get(url, timeout=5) as response: return await response.text() except asyncio.TimeoutError: print(f"{url} 타임아웃") return None except Exception as e: print(f"{url} 에러: {e}") return None

asyncio.gather()는 기본적으로 하나의 작업이 실패하면 전체가 실패합니다. return_exceptions=True 옵션을 추가하면 예외를 결과로 반환받을 수 있어요.

3. CPU 집약적 작업

비동기는 I/O 대기 시간을 활용하는 방식이므로, CPU를 많이 쓰는 계산 작업에는 적합하지 않습니다. 이런 경우엔 multiprocessing이나 concurrent.futures를 사용하세요.

실무 활용 팁¶

비동기 프로그래밍을 실무에 적용할 때 알아두면 좋은 팁들을 공유합니다.

세마포어로 동시 실행 제한하기

python async def main(): semaphore = asyncio.Semaphore(5) # 최대 5개만 동시 실행

async def limited_fetch(url):
    async with semaphore:
        return await fetch_url(session, url)

tasks = [limited_fetch(url) for url in urls]
results = await asyncio.gather(*tasks)

서버에 과부하를 주지 않으려면 동시 요청 수를 제한해야 합니다. Semaphore를 사용하면 간단히 해결돼요.

백그라운드 태스크 생성

python async def background_task(): while True: print("백그라운드 작업 실행") await asyncio.sleep(10)

async def main(): task = asyncio.create_task(background_task()) # 메인 로직 실행 await asyncio.sleep(35) task.cancel() # 종료

create_task()를 사용하면 코루틴을 백그라운드에서 실행하면서 다른 작업을 계속할 수 있습니다.

💡 실무 조언: FastAPI, Sanic 같은 비동기 웹 프레임워크를 사용하면 전체 애플리케이션을 비동기로 구성해 높은 성능을 얻을 수 있습니다.

비동기 프로그래밍은 처음엔 어렵지만, 한 번 익숙해지면 I/O 작업이 많은 프로그램의 성능을 극적으로 향상시킬 수 있는 강력한 도구입니다. 작은 프로젝트부터 시작해서 점차 복잡한 시스템에 적용해보세요!

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