[FastAPI] 비동기로 RabbitMQ 연결하기

🔥 개요

FastAPI와 RabbitMQ를 활용하여 비동기 메시지 소비 시스템을 구축하는 방법을 설명한다. RabbitMQ는 메시지 큐를 제공하여 애플리케이션 간의 비동기 통신을 가능하게 한다. 이를 FastAPI와 결합하면 고성능의 비동기 메시지 처리 서비스를 구현할 수 있다.

이번 포스팅에서는 FastAPI 서버와 RabbitMQ를 연결하는 방법, 메시지를 소비하는 방식, 비동기적으로 외부 프로세스를 실행하는 방법을 간략하게 기술한다.

---

📌 RabbitMQ란?

RabbitMQ는 AMQP(Advanced Message Queuing Protocol) 기반의 메시지 브로커로, 메시지를 큐에 저장하고 비동기적으로 분산 시스템 간 데이터를 전달하는 역할을 한다.

RabbitMQ를 사용하면 다음과 같은 이점이 있다:

• 비동기 메시징: 작업을 다른 서비스로 넘기고 비동기적으로 실행 가능
• 부하 분산: 여러 개의 소비자(Consumer)가 메시지를 나누어 처리 가능
• 장애 대응: 메시지를 큐에 저장하여 장애 발생 시 복구 가능
• 확장성: 다양한 서비스와 통합 가능 (Django, FastAPI, Node.js 등)

---

🛠 RabbitMQ + FastAPI 환경 구축

1️⃣ RabbitMQ 설치 및 실행

RabbitMQ는 Docker를 이용하여 간단히 실행할 수 있다.

docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 \
  -e RABBITMQ_DEFAULT_USER=admin \
  -e RABBITMQ_DEFAULT_PASS=admin \
  rabbitmq:3-management

RabbitMQ 대시보드에 접속하여 큐 상태를 확인할 수 있다.

• URL: http://localhost:15672/
• 로그인 정보: admin / admin

2️⃣ FastAPI 프로젝트 생성

mkdir fastapi-rabbitmq && cd fastapi-rabbitmq
python -m venv venv
source venv/bin/activate  # (Windows의 경우 venv\Scripts\activate)
pip install fastapi uvicorn aio_pika

3️⃣ FastAPI 서버 설정 (main.py)

import asyncio
import signal
from fastapi import FastAPI
from contextlib import asynccontextmanager

IS_PROCESSING = False

def handle_exit_signal(loop):
    def _handler(signal_received, frame):
        print(f"Received exit signal: {signal_received}. Initiating shutdown...")
        loop.create_task(shutdown_handler())

    signal.signal(signal.SIGTERM, _handler)
    signal.signal(signal.SIGINT, _handler)

async def shutdown_handler():
    global IS_PROCESSING
    while IS_PROCESSING:
        print("Waiting for Blender process to complete...")
        await asyncio.sleep(1)
    print("Blender process completed. Shutting down...")

@asynccontextmanager
async def lifespan(app: FastAPI):
    loop = asyncio.get_event_loop()
    handle_exit_signal(loop)
    connection = await connect_to_rabbitmq(loop)
    try:
        yield
    finally:
        await shutdown_handler()
        await connection.close()

app = FastAPI(lifespan=lifespan)

이제 FastAPI가 실행될 때 RabbitMQ와 연결하고, 종료될 때 연결을 닫도록 설정.

---

🏗️ RabbitMQ 연결 및 메시지 소비자 (Consumer) 설정

4️⃣ RabbitMQ 연결 및 메시지 소비자 등록

import asyncio
from aio_pika import connect_robust, IncomingMessage

async def process_message(message: IncomingMessage):
    """RabbitMQ 메시지를 처리하는 비동기 함수"""
    global IS_PROCESSING
    async with processing_lock:
        try:
            IS_PROCESSING = True
            body = message.body.decode()
            print(f"Received message: {body}")
            await message.ack()
        except Exception as e:
            print(f"Error processing message: {e}")
            await message.nack(requeue=True)
        finally:
            IS_PROCESSING = False

async def connect_to_rabbitmq(loop):
    """RabbitMQ 연결 및 소비자 등록"""
    connection = await connect_robust(settings.RABBITMQ_URL, loop=loop)
    channel = await connection.channel()
    await channel.set_qos(prefetch_count=1)
    queue = await channel.declare_queue("task_queue", durable=True)
    await queue.consume(process_message)
    return connection

5️⃣ 메시지 발행자 (Producer) 설정

RabbitMQ 큐에 메시지를 발행하는 코드.

import aio_pika
import asyncio

async def send_message(message: str):
    """RabbitMQ에 메시지를 보내는 함수"""
    connection = await aio_pika.connect_robust("amqp://admin:admin@localhost/")
    async with connection:
        channel = await connection.channel()
        queue = await channel.declare_queue("task_queue", durable=True)
        await channel.default_exchange.publish(
            aio_pika.Message(body=message.encode()),
            routing_key=queue.name
        )

asyncio.run(send_message("Hello, RabbitMQ!"))

---

🎯 비동기적으로 메시지를 처리하는 이유

✅ 블로킹 없이 동시 처리 가능

• aio_pika 라이브러리를 사용하여 RabbitMQ 메시지를 비동기적으로 소비.
• 메시지가 도착하면 즉시 다른 작업을 수행할 수 있어 성능 향상.

✅ 병렬 처리 가능

• 여러 개의 Worker가 동일한 큐를 공유하여 로드 밸런싱 가능.
• 동일한 task_queue에 여러 Consumer를 실행하면 자동으로 분산 처리.

---

🚀 결론

FastAPI와 RabbitMQ를 활용하면 비동기적으로 메시지를 소비하고, 외부 프로세스를 실행하는 강력한 아키텍처를 구축할 수 있다.

이번 글에서 다룬 주요 개념을 정리하면:

• ✅ RabbitMQ 설치 및 FastAPI 연동
• ✅ RabbitMQ Consumer(메시지 소비자) 등록 및 메시지 처리
• ✅ 비동기적으로 메시지를 소비하여 성능 향상
• ✅ Producer를 통해 메시지를 발행하고 Consumer가 처리하도록 구성

도움이 되길 바란다.