[AI 서비스 개발] 비동기, 멀티스레딩, 멀티프로세싱, MSA

1. 비동기란? 

- 하나의 스레드에서 여러 작업을 동시에 실행하고 나중에 결과를 받아 처리할 수 있다.

- I/O 바운드 작업에 적합(네트워크, DB 요청, API 호출 등 대기시간이 긴 작업)

- FastAPI는 내부적으로 이벤트루프를 관리하므로 따로 asyncio.run()을 쓰지 않아도 됨

* def로 정의된 엔드포인트는 기존의 동기 방식으로 실행됨 (이벤트 루프를 사용X, CPU 작업 최적화), 스레드풀에서 실행

a- 멀티스레드 기반 처리(ThreadPoolExecutor에서 실행)

- 관련 함수

비동기 기능	설명
async def	비동기 함수 선언
await	비동기 함수 호출 시 사용
asyncio.run()	비동기 함수 실행
asyncio.gather()	여러개의 비동기 작업 병렬 실행
asuncio.create_task()	백그라운드에서 실행할 작업 예약
asyncio.Queue()	비동기 작업 관리
aiohttp	비동기 http 요청

 

* 동기 함수를 FastAPI에서 비동기처리로 실행시키고 싶으면

- 내 cafe_agent api 호출 시 다른 api가 실행되지 않고, 내 작업이 끝날 때 까지 기다림

=> asyncio.to_thread로 함수를 호출하면 된다.

from fastapi import APIRouter, HTTPException
from app.services.agents.cafe_agent_service import cafe_agent
from app.routers.agents.travel_all_schedule_agent_router import TravelPlanRequest
import asyncio

router = APIRouter()

@router.post("/cafe")
async def cafe_response_test(user_input: TravelPlanRequest):
    """
    카페 정보를 가져오는 엔드포인트.
    - CrewAI 실행 후 일정(JSON) 반환.
    - `asyncio.to_thread()`를 사용해 동기 함수를 비동기적으로 실행.
    """
    try:
        result = await asyncio.to_thread(cafe_agent, user_input.model_dump())  # ✅ 비동기 실행
        return {
            "status": "success",
            "message": "카페 리스트가 생성되었습니다.",
            "data": result.json_dict.get("spots", []),
        }

    except asyncio.CancelledError:
        print("[카페 API] 요청이 취소됨.")
        raise HTTPException(status_code=499, detail="요청이 취소되었습니다.")
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

 

2. 이벤트 루프란?

- 비동기 코드에서 여러 작업을 동시에 실행할 수 있도록 관리하는 핵심 엔진

- 비동기 작업을 큐에 넣음 > 이벤트 루프가 큐에서 작업을 하나씩 꺼내 실행 > 작업이 await을 만나면, 다른 작업을 먼저 실행하고 기다림 > 작업이 끝나면 결과를 반환하고, 다음 작업을 실행

 

3. 멀티 스레딩이란?

- 하나의 프로세스에서 여러 개의 스레드를 실행하는 방식

- CPU 연산(연산, 데이터 처리 등) 을 병렬로 처리해 속도를 높일 수 있다.

- 각 스레드는 독립적으로 실행되지만, 같은 메모리 공간을 공유

- Python의 멀티스레드는 GIL(Global Interpreter Lock)때문에 멀티스레드를 사용해도 한 번에 하나의 스레드만 실행되어 병렬 실행이 불가능해 CPU 연산이 많은 작업에서는 멀티스레드가 성능을 발휘하지 못한다.

- FastAPI에서 uvicorn main:app --workers=N을 실행하면, GIL이 각 프로세스마다 독립적으로 적용되어 병렬 처리 가능

* GIL이란? : 한 번에 하나의 스레드만 실행되도록 제한하는 잠금 장치

import threading
import time

def task(name, delay):
    print(f"{name} 시작")
    time.sleep(delay)
    print(f"{name} 완료")

thread1 = threading.Thread(target=task, args=("Thread 1", 2))
thread2 = threading.Thread(target=task, args=("Thread 2", 1))

thread1.start()
thread2.start()

thread1.join()
thread2.join()

 

 

4. 멀티 프로세싱

- 독립적인 프로세스 여러개 실행 (서로 메모리를 공유하지 않음)

- GIL(Global Interpreter Lock) 영향을 받지 않음, 실제 CPU 병렬 처리 가능

- 각 프로세스가 독립적이므로, 하나가 죽어도 다른 프로세스에는 영향 없음

- 이미지, 영상처리 - GPU 활용 가능 , AI모델 실행(딥러닝) - 여러 코어 활용 , 대량 데이터 연산(병렬연산) - CPU 코어 활용

- uvicorn main:app --workers=4 : 기본적으로 uvicorn은 단일 프로세스로 실행되나, --workers 옵션을 사용하면 멀티프로세스로 실행되어 성능이 향상된다.(4는 프로세스 4개 만들라는 의미)

- 각 워커는 별도의 프로세스로 동작하며, 각 프로세스는 독립적인 이벤트 루프를 가지므로 CPU를 많이 사용하는 작업에도 멀티스레드보다 효율적이다.

* 비동기 API (async def)만 사용하면 이벤트루프로 비동기작업을 효율적을로 처리하므로 --workers를 많이 늘릴 필요가 없다. 오히려 이벤트 루프가 열어 개 생성되어 비효율 적일 수 있다. 워커가 많아지면 메모리 사용량이 증가하고, 프로세스 간 통신 비용도 커지므로 보통 CPU 코어 개수와 비슷한 개수로 설정한다.

import multiprocessing
import time

def task(name):
    print(f"{name} 시작")
    time.sleep(2)
    print(f"{name} 완료")

# 2개의 프로세스 실행
process1 = multiprocessing.Process(target=task, args=("프로세스 1",))
process2 = multiprocessing.Process(target=task, args=("프로세스 2",))

process1.start()
process2.start()

process1.join()
process2.join()

print("모든 작업 완료!")

 

5. MSA(Microservices Architecture)

- 하나의 애플리케이션을 여러 개의 작은 서비스(마이크로서비스)로 나누어 개발하는 아키텍처

- 각 서비스는 독립적으로 실행되며, API 또는 메세지 큐(RabbitMQ, Kafka)를 통해 통신한다

- 서비스 간에 독립성이 높아 유지보수와 확장이 용이하다

- 각 서비스는 독립적인 배포가 가능하다

- 하나의 커다란 애플리케이션으로 구성되면 모놀리식 아키텍쳐라고 함(Monolithic)

- 서비스간의 통신을 효율적으로 처리하기 위해 비동기 방식을 많이 사용한다

- MSA 기반의 웹서비스 예시

1. "유저 서비스" → 회원가입, 로그인 관리
2. "결제 서비스" → 결제 처리
3. "추천 서비스" → 사용자에게 맞는 상품 추천