[Architecture] 메시지 큐로 비동기 시스템 구현하기: Kafka vs RabbitMQ

Client --> Service A --> Service B --> Service C
          (blocking)    (blocking)    (processing)

Producer --> [Message Queue] --> Consumer
  (fire)    (buffer/persist)    (process at own pace)

At-Most-Once:   Producer --> Broker (no retry)
                메시지가 유실될 수 있지만 중복 없음

At-Least-Once:  Producer --> Broker --> ACK --> Retry on failure
                메시지 유실 없지만 중복 가능

Exactly-Once:   Producer --> Broker (idempotent + transaction)
                메시지가 정확히 한 번만 전달

                    +-------------------+
                    |   Kafka Cluster   |
                    |                   |
  Producers ------->|  Broker 1         |-------> Consumers
                    |  Broker 2         |         (Consumer Group A)
                    |  Broker 3         |-------> Consumers
                    |                   |         (Consumer Group B)
                    +-------------------+
                           |
                    +------+------+
                    |  KRaft      |
                    |  Controller |
                    +-------------+

기존 구조:
  Kafka Brokers <---> ZooKeeper Ensemble (별도 클러스터)

KRaft 구조:
  Kafka Brokers (일부가 Controller 역할 겸임)
  - 내부 Raft 합의 프로토콜로 메타데이터 관리
  - 별도 ZooKeeper 클러스터 불필요

from confluent_kafka import Producer

conf = {
    'bootstrap.servers': 'localhost:9092',
    'client.id': 'my-producer',
    'acks': 'all',
    'enable.idempotence': True,
    'max.in.flight.requests.per.connection': 5,
    'retries': 2147483647,
    'linger.ms': 5,
    'batch.size': 16384,
    'compression.type': 'lz4',
}

producer = Producer(conf)

def delivery_callback(err, msg):
    if err:
        print(f"Delivery failed: {err}")
    else:
        print(f"Delivered to {msg.topic()} [{msg.partition()}] @ {msg.offset()}")

# 키 기반 파티셔닝 - 같은 키는 같은 파티션으로
producer.produce(
    topic='orders',
    key='user-123',
    value='{"order_id": "ord-456", "amount": 5000}',
    callback=delivery_callback
)

producer.flush()

Producer 내부 동작:
  Record --> Accumulator --> [Batch] --> Compressor --> Network Send

  linger.ms: 배치 전송 대기 시간 (기본 0ms)
  batch.size: 배치 최대 크기 (기본 16KB)
  compression.type: none / gzip / snappy / lz4 / zstd

Topic: orders (3 partitions)

Consumer Group A:
  Consumer 1 <-- Partition 0
  Consumer 2 <-- Partition 1
  Consumer 3 <-- Partition 2

Consumer Group B:
  Consumer 4 <-- Partition 0, 1, 2

from confluent_kafka import Consumer, KafkaError

conf = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'order-processor',
    'auto.offset.reset': 'earliest',
    'enable.auto.commit': False,  # 수동 커밋 권장
    'max.poll.interval.ms': 300000,
    'session.timeout.ms': 45000,
}

consumer = Consumer(conf)
consumer.subscribe(['orders'])

try:
    while True:
        msg = consumer.poll(timeout=1.0)
        if msg is None:
            continue
        if msg.error():
            if msg.error().code() == KafkaError._PARTITION_EOF:
                continue
            else:
                print(f"Error: {msg.error()}")
                break

        # 메시지 처리
        process_order(msg.value().decode('utf-8'))

        # 수동 커밋 - 처리 완료 후
        consumer.commit(asynchronous=False)
finally:
    consumer.close()

Rebalancing 트리거:
  1. 새로운 소비자 합류
  2. 기존 소비자 이탈 (crash 또는 정상 종료)
  3. 토픽 파티션 수 변경
  4. 구독 토픽 변경

Rebalancing 전략:
  - Eager (Stop-the-World): 모든 파티션 해제 후 재할당
  - Cooperative (Incremental): 변경된 파티션만 재할당

# Idempotent Producer + Transactional Messaging
conf = {
    'bootstrap.servers': 'localhost:9092',
    'enable.idempotence': True,
    'transactional.id': 'my-transaction-id',
}

producer = Producer(conf)
producer.init_transactions()

try:
    producer.begin_transaction()

    producer.produce('topic-a', key='k1', value='v1')
    producer.produce('topic-b', key='k2', value='v2')

    # Consumer offset도 트랜잭션에 포함
    producer.send_offsets_to_transaction(
        consumer.position(consumer.assignment()),
        consumer.consumer_group_metadata()
    )

    producer.commit_transaction()
except Exception as e:
    producer.abort_transaction()
    raise e

Kafka Connect:
  Source Connector: DB/File/API --> Kafka Topic
  Sink Connector:   Kafka Topic --> DB/File/API

  예시: Debezium (MySQL CDC) --> Kafka --> Elasticsearch Sink

Kafka Streams:
  Kafka Topic --> Stream Processing --> Kafka Topic

  특징: 별도 클러스터 불필요, 라이브러리 형태
  기능: filter, map, groupBy, windowed aggregation, join

Producer --> Exchange --> Binding --> Queue --> Consumer

  Exchange: 메시지 라우팅 허브
  Binding:  Exchange와 Queue 간의 라우팅 규칙
  Queue:    메시지가 저장되는 버퍼
  Virtual Host: 논리적 격리 단위 (멀티 테넌트)

Producer --> [Direct Exchange]
              |
              +--(routing_key="order.created")--> [Order Queue]
              |
              +--(routing_key="payment.processed")--> [Payment Queue]

import pika

connection = pika.BlockingConnection(
    pika.ConnectionParameters('localhost')
)
channel = connection.channel()

# Direct Exchange 선언
channel.exchange_declare(
    exchange='orders',
    exchange_type='direct',
    durable=True
)

# Queue 선언 및 바인딩
channel.queue_declare(queue='order_created', durable=True)
channel.queue_bind(
    exchange='orders',
    queue='order_created',
    routing_key='order.created'
)

# 메시지 발행
channel.basic_publish(
    exchange='orders',
    routing_key='order.created',
    body='{"order_id": "123", "user_id": "456"}',
    properties=pika.BasicProperties(
        delivery_mode=2,  # persistent
        content_type='application/json',
    )
)

Producer --> [Topic Exchange]
              |
              +--(routing_key="order.*.kr")--> [Korea Order Queue]
              |
              +--(routing_key="order.premium.#")--> [Premium Queue]
              |
              +--(routing_key="#")--> [Audit Queue] (모든 메시지)

  * : 정확히 하나의 단어 매칭
  # : 0개 이상의 단어 매칭

Producer --> [Fanout Exchange]
              |
              +--> [Queue A] (모든 메시지)
              |
              +--> [Queue B] (모든 메시지)
              |
              +--> [Queue C] (모든 메시지)

def callback(ch, method, properties, body):
    try:
        process_message(body)
        # 처리 성공 시 ACK
        ch.basic_ack(delivery_tag=method.delivery_tag)
    except Exception as e:
        # 처리 실패 시 NACK (requeue)
        ch.basic_nack(
            delivery_tag=method.delivery_tag,
            requeue=True
        )

# Prefetch 설정 - 한 번에 처리할 메시지 수 제한
channel.basic_qos(prefetch_count=10)

channel.basic_consume(
    queue='order_created',
    on_message_callback=callback,
    auto_ack=False  # 수동 ACK
)

channel.start_consuming()

# DLQ 설정
channel.queue_declare(
    queue='order_processing',
    durable=True,
    arguments={
        'x-dead-letter-exchange': 'dlx',
        'x-dead-letter-routing-key': 'order.failed',
        'x-message-ttl': 30000,       # 30초 TTL
        'x-max-length': 10000,         # 최대 큐 길이
    }
)

# DLX (Dead Letter Exchange) 설정
channel.exchange_declare(exchange='dlx', exchange_type='direct')
channel.queue_declare(queue='order_failed', durable=True)
channel.queue_bind(
    exchange='dlx',
    queue='order_failed',
    routing_key='order.failed'
)

channel.confirm_delivery()

try:
    channel.basic_publish(
        exchange='orders',
        routing_key='order.created',
        body=message_body,
        properties=pika.BasicProperties(delivery_mode=2),
        mandatory=True  # 라우팅 실패 시 반환
    )
    print("Message confirmed by broker")
except pika.exceptions.UnroutableError:
    print("Message could not be routed")
except pika.exceptions.NackError:
    print("Message was nacked by broker")

Classic Mirrored Queue (레거시):
  Node 1 (Leader)  <-mirror-> Node 2 (Mirror) <-mirror-> Node 3 (Mirror)
  문제: split-brain 위험, 동기화 오버헤드

Quorum Queue (권장):
  Node 1 (Leader)  <-Raft-> Node 2 (Follower) <-Raft-> Node 3 (Follower)
  장점: Raft 합의 프로토콜, split-brain 방지, 데이터 안전성 보장

channel.queue_declare(
    queue='important_orders',
    durable=True,
    arguments={
        'x-queue-type': 'quorum',
        'x-quorum-initial-group-size': 3,
    }
)

channel.queue_declare(
    queue='order_stream',
    durable=True,
    arguments={
        'x-queue-type': 'stream',
        'x-max-length-bytes': 1073741824,  # 1GB
        'x-stream-max-segment-size-bytes': 52428800,  # 50MB
    }
)

+----------+     +-------------+     +----------+     +--------------+
|          |     |             |     |          |     |              |
|  Client  +---->+ FastAPI     +---->+ Request  +---->+ Worker Pool  |
|  (Web)   |     | Gateway     |     | Queue    |     | (Celery)     |
|          |     |             |     | (Redis)  |     |              |
+----------+     +------+------+     +----------+     +------+-------+
                        |                                     |
                        |            +----------+             |
                        |            |          |             |
                        +<-----------+ Response +<------------+
                        |            | Store    |
                        |            | (Redis)  |
                        |            +----------+
                        v
                  +----------+
                  |  Status  |
                  |  SSE/WS  |
                  +----------+

from enum import IntEnum

class Priority(IntEnum):
    CRITICAL = 0   # 실시간 챗봇 응답
    HIGH = 1       # 프리미엄 사용자
    NORMAL = 2     # 일반 사용자
    LOW = 3        # 배치 처리
    BACKGROUND = 4 # 비실시간 분석

# Celery Task with Priority
from celery import Celery

app = Celery('llm_gateway', broker='redis://localhost:6379/0')

@app.task(bind=True, max_retries=3, default_retry_delay=60)
def call_llm(self, request_id, model, messages, priority):
    try:
        response = litellm.completion(
            model=model,
            messages=messages,
            timeout=30,
        )
        store_response(request_id, response)
        return response
    except Exception as exc:
        # Exponential backoff
        retry_delay = 60 * (2 ** self.request.retries)
        raise self.retry(exc=exc, countdown=retry_delay)

import asyncio
import random

async def call_llm_with_retry(
    model: str,
    messages: list,
    max_retries: int = 5,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
):
    for attempt in range(max_retries + 1):
        try:
            response = await litellm.acompletion(
                model=model,
                messages=messages,
                timeout=30,
            )
            return response
        except Exception as e:
            if attempt == max_retries:
                raise

            # Exponential backoff + jitter
            delay = min(base_delay * (2 ** attempt), max_delay)
            jitter = random.uniform(0, delay * 0.1)
            actual_delay = delay + jitter

            print(f"Attempt {attempt + 1} failed: {e}")
            print(f"Retrying in {actual_delay:.1f}s...")
            await asyncio.sleep(actual_delay)

# app/main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from celery.result import AsyncResult
import uuid

app = FastAPI()

class LLMRequest(BaseModel):
    model: str = "gpt-4o"
    messages: list
    priority: int = 2

class LLMResponse(BaseModel):
    request_id: str
    status: str
    result: dict = None

@app.post("/api/v1/completions")
async def create_completion(req: LLMRequest):
    request_id = str(uuid.uuid4())

    # 우선순위에 따라 Celery 큐에 전송
    task = call_llm.apply_async(
        args=[request_id, req.model, req.messages, req.priority],
        queue=f"llm_priority_{req.priority}",
        priority=req.priority,
    )

    return {
        "request_id": request_id,
        "task_id": task.id,
        "status": "queued",
    }

@app.get("/api/v1/completions/{task_id}")
async def get_completion(task_id: str):
    result = AsyncResult(task_id)

    if result.ready():
        return {
            "status": "completed",
            "result": result.get(),
        }
    elif result.failed():
        return {
            "status": "failed",
            "error": str(result.result),
        }
    else:
        return {
            "status": "processing",
        }

# celery_config.py
from celery import Celery

app = Celery('llm_gateway')

app.conf.update(
    broker_url='redis://localhost:6379/0',
    result_backend='redis://localhost:6379/1',
    task_serializer='json',
    result_serializer='json',
    accept_content=['json'],
    timezone='UTC',
    task_routes={
        'tasks.call_llm': {
            'queue': 'llm_default',
        },
    },
    # 우선순위 큐 설정
    task_queue_max_priority=10,
    task_default_priority=5,
    # Worker 동시성 제한 (Rate Limit 대응)
    worker_concurrency=4,
    # 태스크 시간 제한
    task_soft_time_limit=120,
    task_time_limit=180,
)

# docker-compose.yml
version: '3.8'
services:
  api:
    build: .
    command: uvicorn app.main:app --host 0.0.0.0 --port 8000
    ports:
      - '8000:8000'
    depends_on:
      - redis
    environment:
      - REDIS_URL=redis://redis:6379

  worker-high:
    build: .
    command: celery -A tasks worker -Q llm_priority_0,llm_priority_1 -c 4
    depends_on:
      - redis

  worker-normal:
    build: .
    command: celery -A tasks worker -Q llm_priority_2 -c 8
    depends_on:
      - redis

  worker-low:
    build: .
    command: celery -A tasks worker -Q llm_priority_3,llm_priority_4 -c 2
    depends_on:
      - redis

  redis:
    image: redis:7-alpine
    ports:
      - '6379:6379'

  flower:
    build: .
    command: celery -A tasks flower --port=5555
    ports:
      - '5555:5555'
    depends_on:
      - redis

테스트 환경: 3 broker/node, 3 replica, 메시지 크기 1KB

Apache Kafka:
  - 단일 파티션: ~50,000 msg/sec
  - 12 파티션:   ~800,000 msg/sec
  - 60 파티션:   ~2,000,000+ msg/sec
  - 지연시간 p99: 5ms

RabbitMQ (Quorum Queue):
  - 단일 큐:     ~20,000 msg/sec
  - 10 큐:       ~100,000 msg/sec
  - 지연시간 p99: 1ms

소규모/중규모 (일 10만 요청 이하):
  RabbitMQ + Celery 조합 권장
  - 설정 간편, Priority Queue 지원 우수
  - Dead Letter Queue로 실패 처리 용이

대규모 (일 100만 요청 이상):
  Kafka 기반 구현 권장
  - 높은 처리량, 메시지 보존으로 재처리 가능
  - Consumer Group으로 유연한 스케일링
  - 비용/사용량 분석을 위한 이벤트 로그 활용

# Prometheus JMX Exporter 설정 (prometheus-jmx-config.yaml)
rules:
  - pattern: 'kafka.server<type=BrokerTopicMetrics, name=MessagesInPerSec>'
    name: kafka_server_messages_in_total
    type: COUNTER
  - pattern: 'kafka.server<type=BrokerTopicMetrics, name=BytesInPerSec>'
    name: kafka_server_bytes_in_total
    type: COUNTER
  - pattern: 'kafka.consumer<type=consumer-fetch-manager-metrics, client-id=(.+)><>records-lag-max'
    name: kafka_consumer_lag_max
    type: GAUGE

핵심 지표:
  - Queue Depth: 큐에 쌓인 메시지 수
  - Consumer Utilization: 소비자 활용률
  - Publish/Deliver Rate: 초당 발행/전달 메시지 수
  - Unacked Messages: ACK 대기 중인 메시지 수
  - Memory/Disk Usage: 리소스 사용량

RabbitMQ Management UI: http://localhost:15672
Prometheus Plugin: rabbitmq_prometheus (내장)