Python asyncio를 통한 비동기 프로그래밍
학습 목표 매핑
SKALA 3기 Module 1 — Python AI Native (Learning Objective 1-2)
- Objective: async/await 문법으로 비동기 프로그램을 작성하고, 파일 I/O·DB 조회 등 대기 작업에 적용하여 동기 대비 처리량 2배 이상 증가 (Bloom L2-L3)
- Evaluation: 성능 벤치마크 (동기 vs 비동기 실행 시간 비교)
비동기 프로그래밍의 핵심 개념
정의
비동기 프로그래밍은 네트워크 통신이나 파일 입출력 같은 대기 시간을 활용하여 “CPU가 다른 처리를 할 수 있도록” 하는 방식입니다.
전통적인 멀티스레드 방식보다 안전하고 효율적합니다.
동기 vs 비동기 — 성능 비교
동기 처리 (순차 실행 - 6초 소요):
import time
def find_users_sync(n):
for i in range(1, n + 1):
time.sleep(1) # 대기 시간 동안 CPU 낭비
print(f"User {i} found")
# 3개 작업 실행
find_users_sync(1) # 1초
find_users_sync(1) # 1초
find_users_sync(1) # 1초
# 총 3초 (순차)비동기 처리 (병렬 실행 - 1초 소요):
import asyncio
async def find_users_async(n):
for i in range(1, n + 1):
await asyncio.sleep(1) # 대기 중 다른 작업 수행
print(f"User {i} found")
async def process_async():
# 3개 작업 동시 실행
await asyncio.gather(
find_users_async(1), # 1초
find_users_async(1), # 1초 (동시에 실행)
find_users_async(1), # 1초 (동시에 실행)
)
# 총 1초 (병렬 처리)
asyncio.run(process_async())성능 향상: 3초 → 1초 (3배 빠름)
async/await 문법
기본 구조
| 방식 | 호출 | 특징 |
|---|---|---|
def 함수 | 일반 호출 | 동기 방식, 즉시 실행 |
async def 함수 | await로 호출 | 비동기 함수(코루틴), 대기 필요 |
함수 정의
# 동기 함수
def sync_function():
return "result"
# 비동기 함수 (코루틴)
async def async_function():
await asyncio.sleep(1)
return "result"함수 호출
# 동기 함수 호출
result = sync_function() # 즉시 실행, 결과 반환
# 비동기 함수 호출 (일반 호출)
coroutine = async_function() # 코루틴 객체만 반환, 실행 안 됨
# 비동기 함수 호출 (await)
result = await async_function() # 실행 + 대기
# 비동기 함수 실행 시작점 (Python 3.7+)
asyncio.run(async_function())실제 코드 예시
asyncio.gather()로 병렬 실행
import asyncio
import time
async def fetch_data(id, delay):
"""네트워크 요청 시뮬레이션"""
print(f"[{id}] 데이터 요청 시작")
await asyncio.sleep(delay) # 네트워크 대기
print(f"[{id}] 데이터 수신 완료")
return f"data_{id}"
async def main():
# 3개 요청 동시 실행
start = time.time()
results = await asyncio.gather(
fetch_data(1, 2), # 2초
fetch_data(2, 3), # 3초
fetch_data(3, 1), # 1초
)
elapsed = time.time() - start
print(f"결과: {results}")
print(f"소요 시간: {elapsed:.1f}초 (동기는 6초 소요)")
asyncio.run(main())출력:
[1] 데이터 요청 시작
[2] 데이터 요청 시작
[3] 데이터 요청 시작
[3] 데이터 수신 완료
[1] 데이터 수신 완료
[2] 데이터 수신 완료
결과: ['data_1', 'data_2', 'data_3']
소요 시간: 3.0초 (동기는 6초 소요)
성능: 3초 vs 6초 (2배 빠름)
asyncio의 핵심 함수
| 함수 | 역할 | 사용 예 |
|---|---|---|
| asyncio.run() | 비동기 함수 실행 시작점 | asyncio.run(main()) |
| asyncio.gather() | 여러 코루틴 동시 실행 + 결과 수집 | await asyncio.gather(task1, task2, ...) |
| asyncio.create_task() | 백그라운드 작업 생성 | task = asyncio.create_task(coro) |
| asyncio.sleep() | 비동기 대기 (CPU 낭비 안 함) | await asyncio.sleep(1) |
| asyncio.wait() | 여러 작업 대기 (완료까지) | await asyncio.wait(tasks) |
실무 사용 사례
1. 웹 크롤링 (API 여러 개 병렬 호출)
import aiohttp
import asyncio
async def fetch_url(session, url):
async with session.get(url) as response:
return await response.text()
async def scrape_multiple():
urls = [
'https://example.com/1',
'https://example.com/2',
'https://example.com/3',
]
async with aiohttp.ClientSession() as session:
results = await asyncio.gather(*[
fetch_url(session, url) for url in urls
])
return results2. 데이터베이스 쿼리
import asyncpg
async def fetch_users():
conn = await asyncpg.connect('postgresql://...')
# 여러 쿼리 병렬 실행
results = await asyncio.gather(
conn.fetch('SELECT * FROM users WHERE id = 1'),
conn.fetch('SELECT * FROM users WHERE id = 2'),
conn.fetch('SELECT * FROM orders WHERE user_id = 1'),
)
await conn.close()
return results3. FastAPI 비동기 핸들러
from fastapi import FastAPI
app = FastAPI()
@app.get("/items/")
async def get_items():
# 비동기 DB 조회, API 호출 등
data = await fetch_from_db()
return data효과적인 사용 패턴
| 패턴 | 설명 | 예 |
|---|---|---|
| 병렬 처리 | 여러 I/O 작업 동시 실행 | API 여러 개 호출 |
| 백그라운드 작업 | 메인 작업과 독립적 실행 | 로그 기록, 이메일 발송 |
| 캐싱 | 이전 결과 재사용 | 자주 요청되는 데이터 |
주의점
❌ 비동기는 CPU 집약 작업에는 부적합:
# 나쁜 예: CPU 집약적
async def cpu_heavy():
for i in range(10**8): # 계산만 함
x = i ** 2✅ 비동기는 I/O 대기 작업에 최적:
# 좋은 예: I/O 대기
async def io_heavy():
await asyncio.sleep(1) # 네트워크 대기
result = await fetch_from_api() # DB/API 조회학습 설계 포인트
Cognitive Level (Bloom)
- L2 (Understand): async/await 개념 및 코루틴 이해
- L3 (Apply): asyncio 함수를 이용한 비동기 코드 작성
권장 실습
- 개념 이해: 동기 vs 비동기 실행 시간 비교 (벤치마크)
- 코드 작성: 간단한 비동기 함수 작성 (asyncio.sleep 사용)
- 병렬 처리: asyncio.gather()로 여러 작업 동시 실행
- 성능 측정: 동기 → 비동기 리팩토링 후 성능 비교
참고 자료
- Python asyncio 공식: https://docs.python.org/ko/3/library/asyncio.html
- Dale Seo 블로그: https://daleseo.com/python-asyncio/
- Real Python: https://realpython.com/async-io-python/
타 소스와의 연계
python-type-hints-fastapi (비동기 함수 타입 힌트) python-venv-poetry-conda-leapcell (프로젝트 환경에서 비동기 코드 실행)