[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)
  问题: 存在脑裂风险，同步开销大

Quorum Queue (推荐):
  Node 1 (Leader)  <-Raft-> Node 2 (Follower) <-Raft-> Node 3 (Follower)
  优点: Raft 共识协议，防止脑裂，保证数据安全

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 (内置)