파이썬 비동기 코드를 작성하다 보면 'RuntimeError: This event loop is already running' 에러를 마주하게 됩니다. 특히 Jupyter 노트북이나 FastAPI 환경에서 asyncio.run()을 호출할 때 자주 발생하는 이 문제는 이미 실행 중인 이벤트 루프와 새로 생성하려는 루프가 충돌하기 때문입니다. 이 글에서는 이벤트 루프 충돌의 원인을 진단하고, nest_asyncio를 활용한 실전 해결법과 비동기 프로그래밍의 핵심 개념을 코드 중심으로 정리합니다.
 |
| PythonRuntimeErrorThiseventloopisalreadyrunning비동기충돌완벽해결가이드 |
1. 문제 진단: 이벤트 루프 충돌이 발생하는 이유
asyncio는 단일 스레드에서 하나의 이벤트 루프만 실행할 수 있도록 설계되었습니다. 그러나 다음과 같은 환경에서는 이미 이벤트 루프가 백그라운드에서 동작하고 있습니다.
- Jupyter Notebook / IPython: 셀 실행을 위해 내부적으로 이벤트 루프가 항상 실행 중입니다.
- FastAPI / Uvicorn: ASGI 서버가 자체 이벤트 루프를 관리합니다.
- GUI 프레임워크: Tkinter, PyQt 등도 메인 루프를 점유합니다.
이런 환경에서 asyncio.run()을 호출하면 새로운 이벤트 루프를 생성하려 시도하므로 충돌이 발생합니다. asyncio.run()은 내부적으로 새 루프를 만들고(new_event_loop) 종료 시 닫는(close) 구조이기 때문입니다.
2. 해결 방법 1: nest_asyncio로 중첩 루프 허용하기
가장 빠르고 실용적인 해결책은 nest_asyncio 라이브러리를 사용하는 것입니다. 이 라이브러리는 기존 실행 중인 이벤트 루프 위에 새로운 루프를 중첩 실행할 수 있도록 asyncio의 동작을 패치합니다.
# 설치
!pip install nest_asyncio
# 사용법
import nest_asyncio
import asyncio
nest_asyncio.apply()
async def main():
print("이벤트 루프 충돌 없이 실행됩니다.")
await asyncio.sleep(1)
return "완료"
# Jupyter 환경에서도 정상 동작
result = asyncio.run(main())
print(result)
핵심 포인트:
- nest_asyncio.apply()는 프로그램 시작 시 한 번만 호출하면 됩니다.
- 성능 오버헤드는 거의 없으며, 개발 환경에서의 편의성을 위한 도구입니다.
- 프로덕션 코드보다는 노트북 기반 프로토타이핑이나 디버깅 용도로 권장합니다.
3. 해결 방법 2: 환경별 올바른 비동기 실행 패턴
nest_asyncio 없이 환경에 맞는 실행 방식을 사용하는 것이 더 근본적인 해결책입니다.
3-1. Jupyter Notebook 환경
Jupyter는 이미 실행 중인 루프가 있으므로 Top-level await를 사용합니다.
# asyncio.run() 대신 직접 await 사용
async def fetch_data():
await asyncio.sleep(2)
return "데이터 로드 완료"
# 노트북 셀에서 바로 실행
result = await fetch_data()
print(result)
3-2. 일반 Python 스크립트
스크립트 환경에서는 asyncio.run()이 표준 패턴입니다.
# main.py
import asyncio
async def main():
tasks = [
asyncio.create_task(process(i))
for i in range(5)
]
results = await asyncio.gather(*tasks)
return results
if __name__ == "__main__":
asyncio.run(main())
3-3. FastAPI 내부에서 비동기 함수 호출
FastAPI는 자체 이벤트 루프를 관리하므로 await만 사용합니다.
from fastapi import FastAPI
app = FastAPI()
@app.get("/data")
async def get_data():
# asyncio.run() 사용 금지
result = await fetch_from_db()
return {"data": result}
4. 비동기 프로그래밍 핵심 개념 정리
이벤트 루프 충돌을 이해하려면 비동기의 기본 원리를 알아야 합니다.
4-1. async/await와 코루틴
- async def: 코루틴 함수를 정의합니다. 호출 시 즉시 실행되지 않고 코루틴 객체를 반환합니다.
- await: 코루틴이 완료될 때까지 제어권을 이벤트 루프에 반환합니다. 이때 함수 내부 상태(지역 변수 등)는 유지됩니다.
- 중단/재개: 일반 함수와 달리 await 지점에서 멈췄다가 작업 완료 후 다시 이어집니다.
4-2. 동시 실행을 위한 태스크 관리
단순히 await를 나열하면 순차 실행됩니다. 진짜 동시성을 얻으려면:
# 잘못된 예: 순차 실행 (총 5초 소요) await asyncio.sleep(3) await asyncio.sleep(2) # 올바른 예: 동시 실행 (총 3초 소요) task1 = asyncio.create_task(asyncio.sleep(3)) task2 = asyncio.create_task(asyncio.sleep(2)) await asyncio.gather(task1, task2)
- asyncio.create_task(): 코루틴을 태스크로 등록하여 이벤트 루프에 예약합니다.
- asyncio.gather(): 여러 코루틴/태스크를 동시 실행하고 모든 결과를 수집합니다.
4-3. 블로킹 함수 주의사항
time.sleep()은 절대 사용하지 마십시오. 이 함수는 전체 스레드를 블로킹하여 이벤트 루프를 멈춥니다.
# 잘못된 코드: 이벤트 루프 블로킹
import time
async def wrong():
time.sleep(1) # 다른 모든 태스크도 멈춤
# 올바른 코드: 논블로킹
async def correct():
await asyncio.sleep(1) # 다른 태스크는 계속 실행됨
4-4. 동시성(Concurrency) vs 병렬성(Parallelism)
- 동시성: 단일 코어에서 여러 작업을 시간 분할하여 번갈아 수행합니다. asyncio가 이에 해당합니다.
- 병렬성: 여러 코어에서 작업을 물리적으로 동시 수행합니다. multiprocessing이 이에 해당합니다.
- asyncio는 I/O bound 작업(네트워크, 파일 읽기 등)에 효율적이며, CPU bound 작업에는 적합하지 않습니다.
5. 실전 트러블슈팅 체크리스트
비동기 코드 작성 시 다음 사항을 확인하십시오.
- 환경 확인: Jupyter인가 스크립트인가? → 실행 방식(await vs asyncio.run) 결정
- 이벤트 루프 중복: 이미 실행 중인 루프가 있는가? → nest_asyncio 적용 고려
- 태스크 등록: create_task 또는 gather를 사용했는가? → 동시 실행 보장
- 블로킹 함수: time.sleep, requests 같은 동기 함수를 사용하지 않았는가? → asyncio.sleep, aiohttp로 대체
- 예외 처리: 태스크 내부 예외는 gather의 return_exceptions=True로 처리
마무리
이벤트 루프 충돌은 asyncio의 단일 루프 정책과 환경별 루프 관리 방식의 차이에서 발생합니다. nest_asyncio는 빠른 우회 수단이지만, 근본적으로는 환경에 맞는 실행 패턴(Jupyter에서는 await, 스크립트에서는 asyncio.run)을 사용하는 것이 권장됩니다. 비동기 프로그래밍의 핵심은 await로 제어권을 넘기고, create_task/gather로 동시성을 확보하며, 블로킹 함수를 철저히 배제하는 것입니다.
Action Item: 지금 당장 당신의 코드에서 time.sleep을 검색하여 asyncio.sleep으로 교체하고, Jupyter 환경이라면 파일 최상단에 nest_asyncio.apply()를 추가하십시오.
# 함께 보면 좋은 글
댓글
댓글 쓰기