RabbitMQ on Kubernetes 面试题库

Producer → Exchange → [Binding] → Queue → Consumer
                ↑                    ↑
           Channel              Channel
                ↑                    ↑
           Connection           Connection

Exchange: order.exchange (type=direct)

Binding: order.exchange → queue.order.create  (routing_key: order.create)
Binding: order.exchange → queue.order.cancel  (routing_key: order.cancel)
Binding: order.exchange → queue.order.pay     (routing_key: order.pay)

消息 routing_key=order.pay → 仅路由到 queue.order.pay

Exchange: log.exchange (type=topic)

Binding: log.exchange → queue.error    (routing_key: *.error.*)
Binding: log.exchange → queue.app1     (routing_key: app1.#)
Binding: log.exchange → queue.all      (routing_key: #)

消息 routing_key=app1.error.db → 路由到 queue.error、queue.app1、queue.all
消息 routing_key=app2.info      → 仅路由到 queue.all

Exchange: cache.invalidate (type=fanout)

Binding: cache.invalidate → queue.service-a
Binding: cache.invalidate → queue.service-b
Binding: cache.invalidate → queue.service-c

任意消息 → 同时投递到三个队列

1. channel.ConfirmSelect()        // 开启发布确认模式
2. channel.BasicPublish(...)       // 发布消息
3. 异步等待：
   - BasicAck(deliveryTag, multiple)   // 成功确认
   - BasicNack(deliveryTag, multiple)  // 失败通知
4. 未确认消息重试或记录异常

# 声明持久化队列
durable: true

# 发送持久化消息（Delivery Mode = 2）
props := amqp.Table{
    "delivery_mode": 2,
}

graph TD
    K8sAPI["Kubernetes API Server"] --> Operator["RabbitMQ Cluster Operator"]
    Operator --> Controller["Controller (Reconcile Loop)"]
    Controller --> |监听| CR["RabbitmqCluster CR"]
    Controller --> |创建/更新| STS["StatefulSet"]
    Controller --> |创建| SVC["Service / ConfigMap"]
    Controller --> |管理| SECRET["Secret (证书/凭据)"]
    Controller --> |处理| ROLLING["滚动升级"]
    Operator --> Cluster["RabbitMQ Cluster (StatefulSet)"]
    Cluster --> Pod0["Pod-0"]
    Cluster --> Pod1["Pod-1"]
    Cluster --> Pod2["Pod-2"]

apiVersion: rabbitmq.com/v1beta1
kind: RabbitmqCluster
metadata:
  name: production-cluster
spec:
  replicas: 3
  image: rabbitmq:3.12-management
  service:
    type: ClusterIP
  persistence:
    storageClassName: ssd
    storage: 100Gi
  resources:
    requests:
      cpu: "2"
      memory: 4Gi
    limits:
      cpu: "4"
      memory: 8Gi
  rabbitmq:
    additionalConfig: |
      vm_memory_high_watermark.relative = 0.6
      disk_free_limit.relative = 2.0      

graph TD
    Headless["Headless Service: rabbitmq-nodes<br/>clusterIP: None<br/>DNS: rabbitmq-nodes.namespace.svc.cluster.local"] --> Pod0["Pod-0<br/>rabbit@rmq-0<br/>PVC-0"]
    Headless --> Pod1["Pod-1<br/>rabbit@rmq-1<br/>PVC-1"]
    Headless --> Pod2["Pod-2<br/>rabbit@rmq-2<br/>PVC-2"]

    ClientSvc["Client Service: rabbitmq<br/>clusterIP: allocated<br/>端口 5672/15672"]

apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: rabbitmq
spec:
  serviceName: rabbitmq-nodes       # Headless Service
  replicas: 3
  podManagementPolicy: Parallel     # 并行启动，加速集群形成
  updateStrategy:
    type: RollingUpdate
  volumeClaimTemplates:             # 每 Pod 独立 PVC
  - metadata:
      name: data
    spec:
      accessModes: ["ReadWriteOnce"]
      storageClassName: ssd
      resources:
        requests:
          storage: 100Gi

graph TD
    C["Client"] --> L["Leader (Raft Node)"]
    L --> F1["Follower (ACK)"]
    L --> F2["Follower (ACK)"]
    L --> F3["Follower (ACK)"]

# 创建 Stream
rabbitmqadmin declare queue name=my.stream durable=true arguments='{"x-queue-type":"stream"}'

# 消费者从指定 Offset 开始消费
# 使用 RabbitMQ Stream Client（Java/Go/.NET）

1. Pod 重启，PVC 重新挂载
2. RabbitMQ 进程启动，加载 Mnesia 数据库
3. 恢复 Exchange / Queue / Binding 定义
4. 恢复持久化消息（Message Store 重放）
5. 非持久化消息 — 丢弃
6. 重新加入集群（节点发现 → Erlang Cookie 验证 → 数据同步）

spec:
  persistence:
    storageClassName: premium-rwo    # 高性能块存储
    storage: 200Gi                   # 建议留有 2-3 倍余量
  override:
    rabbitmq:
      spec:
        containers:
        - name: rabbitmq
          volumeMounts:
          - name: persistence
            mountPath: /var/lib/rabbitmq/mnesia

# 方式一：声明队列时指定 DLX
arguments:
  x-dead-letter-exchange: "order.dlx"
  x-dead-letter-routing-key: "order.dlq"

# 方式二：通过 Policy 全局控制
rabbitmqctl set_policy DLX ".*" '{
  "dead-letter-exchange": "dead.letter.exchange"
}' --apply-to queues

graph TD
    A["Producer"]
    B["Normal Exchange"]
    C["Normal Queue"]
    D["Consumer<br/>(NACK, requeue=false)"]
    E["DLX Exchange"]
    F["Dead Letter Queue"]
    G["告警 / 人工处理"]
    A --> B --> C --> D
    D -->|NACK| E --> F --> G

# Policy 方式配置
rabbitmqctl set_policy TTL "orders.*" '{
  "message-ttl": 86400000
}' --apply-to queues

# 队列声明时配置
arguments:
  x-message-ttl: 60000         # 队列内消息 60 秒过期
  x-expires: 3600000           # 队列空闲 1 小时后自动删除
  x-dead-letter-exchange: dlx  # 过期消息发送到 DLX

graph TD
    P["Producer"] --> TCP["TCP Connection"] --> B["Broker"]
    B --> CHECK{"检查资源状态"}
    CHECK -->|"资源充足"| NORMAL["正常接收消息"]
    CHECK -->|"达到阈值"| BLOCK["发送 Channel.Blocking<br/>阻断 Producer 发送<br/>拒绝新连接（可选）"]

# rabbitmq.conf
vm_memory_high_watermark.relative = 0.6           # K8s limits 通常低于物理机，可适度放宽
disk_free_limit.relative = 2.0                    # 相对值模式
total_memory_available_override_value = 8589934592 # 显式声明容器可用内存（8G）

graph TD
    RMQ["RabbitMQ Pod (Port 15692)<br/>rabbitmq_prometheus Plugin<br/>→ /metrics (Prometheus text format)"]
    Prom["Prometheus<br/>- ServiceMonitor / PodMonitor<br/>- Scrape Interval: 15s-30s"]
    Grafana["Grafana<br/>- Dashboard: RabbitMQ-Overview<br/>- Dashboard ID: 10991 / 11347"]
    RMQ --> Prom --> Grafana

apiVersion: monitoring.coreos.com/v1
kind: ServiceMonitor
metadata:
  name: rabbitmq
spec:
  selector:
    matchLabels:
      app.kubernetes.io/name: rabbitmq
  endpoints:
  - port: prometheus
    interval: 30s
    path: /metrics

graph LR
    subgraph 多数派["多数派（正常运行）"]
        P0["P0"]
        P1["P1"]
    end
    subgraph 少数派["少数派（自动暂停）"]
        P2["P2"]
    end

分区前：3节点集群
分区后：[P0, P1] vs [P2]
分区恢复后：选择节点数多的分区获胜（[P0, P1]），[P2]清空数据并重新同步

# rabbitmq.conf
cluster_partition_handling = pause_minority
cluster_keepalive_interval = 10000

graph TD
    P["Producer"] --> L["Leader (Raft Node)"]
    L -->|"1. 追加 Entry 到本地 Raft Log"| LOG["Raft Log"]
    L -->|"2. 并行复制到 Follower"| F1["Follower-1 (ACK)"]
    L --> F2["Follower-2 (ACK)"]
    L --> F3["Follower-3 (未响应)"]

# 声明队列时设置最大优先级
arguments:
  x-max-priority: 10    # 0-255，推荐不超过 10

// Go AMQP Client
amqp.Publishing{
    Priority:    9,    // 0 为默认（最低），数字越大优先级越高
    ContentType: "text/plain",
    Body:        []byte("high priority message"),
}

graph TD
    A["Producer"]
    B["Delayed Exchange<br/>(type=x-delayed-message)<br/>消息携带 x-delay header<br/>Exchange 内部暂存<br/>延迟到期后按原 Routing Key 路由"]
    C["Target Queue"]
    D["Consumer"]
    A --> B --> C --> D

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

# rabbitmqadmin 声明
rabbitmqadmin declare exchange name=order.delayed \
  type=x-delayed-message \
  arguments='{"x-delayed-type":"direct"}'

headers := amqp.Table{
    "x-delay": 30000,       // 延迟 30 秒
}
amqp.Publishing{
    Headers: headers,
    Body:    []byte("delayed message"),
}

graph LR
    Upstream["Upstream 集群 (源)<br/>Exchange: order (Federated)"] -->|"Federation Link"| Downstream["Downstream 集群 (目标)<br/>Exchange: order (本地)"]

# 1. 设置 Upstream 连接参数
rabbitmqctl set_parameter federation-upstream upstream-cluster '
{
  "uri": "amqps://remote-cluster:5671",
  "expires": 3600000
}'

# 2. 设置 Policy 将 Federation 应用于 Exchange
rabbitmqctl set_policy federate-me "^order\." '
{
  "federation-upstream": "upstream-cluster"
}' --apply-to exchanges

graph LR
    Source["Source Broker<br/>Queue: A"] --> Shovel["Shovel Plugin<br/>内部进程"] --> Dest["Destination Broker<br/>Exchange: B"]

    Confirm["确认模式：<br/>on-confirm — 源 Broker ACK 后才从源队列删除<br/>on-publish — 发送到目标 Broker 即删除源消息"]

rabbitmqctl set_parameter shovel my-shovel '
{
  "src-uri": "amqp://source-cluster",
  "src-queue": "orders.backlog",
  "dest-uri": "amqp://target-cluster",
  "dest-exchange": "orders.process",
  "ack-mode": "on-confirm",
  "delete-after": "queue-length"
}'

apiVersion: rabbitmq.com/v1beta1
kind: RabbitmqCluster
metadata:
  name: tls-cluster
spec:
  tls:
    secretName: rabbitmq-tls-secret     # 包含 tls.crt / tls.key / ca.crt
    caSecretName: ca-cert-secret        # CA 证书（客户端证书验证时必需）
    disableNonTLSListeners: true        # 禁用非 TLS 端口
  rabbitmq:
    additionalConfig: |
      ssl_options.verify = verify_peer
      ssl_options.fail_if_no_peer_cert = true
      ssl_options.versions.1 = tlsv1.3
      ssl_options.versions.2 = tlsv1.2      

apiVersion: v1
kind: Secret
metadata:
  name: rabbitmq-tls-secret
type: kubernetes.io/tls
data:
  tls.crt: <base64-encoded-certificate>
  tls.key: <base64-encoded-private-key>
apiVersion: v1
kind: Secret
metadata:
  name: ca-cert-secret
data:
  ca.crt: <base64-encoded-ca-certificate>

graph TD
    subgraph BROKER["RabbitMQ Broker"]
        subgraph VHOST1["vhost: / (默认)"]
            U1["User"] --- R1["Role"] --- P1["Perm"]
        end
        subgraph VHOST2["vhost: app-a"]
            U2["User"] --- R2["Role"] --- P2["Perm"]
        end
    end

# 创建用户
rabbitmqctl add_user app_user secure_password

# 设置角色
rabbitmqctl set_user_tags app_user none

# 授予 vhost 权限
rabbitmqctl set_permissions -p vhost_app \
  app_user "^app\." "^app\." "^app\."      # configure write read

# 在 K8s 中通过 CR 管理额外用户

spec:
  rabbitmq:
    additionalConfig: |
      default_user = admin
      default_pass = admin123    # 仅开发环境      
  secretBackend:
    externalSecret:
      name: rabbitmq-credentials  # 外部 Secret 管理凭据

spec:
  affinity:
    podAntiAffinity:
      requiredDuringSchedulingIgnoredDuringExecution:
      - labelSelector:
          matchLabels:
            app.kubernetes.io/name: rabbitmq
        topologyKey: kubernetes.io/hostname

spec:
  affinity:
    podAntiAffinity:
      requiredDuringSchedulingIgnoredDuringExecution:
      - labelSelector:
          matchLabels:
            app.kubernetes.io/name: rabbitmq
        topologyKey: topology.kubernetes.io/zone

spec:
  tolerations:
  - key: "dedicated"
    operator: "Equal"
    value: "rabbitmq"
    effect: "NoSchedule"

# 导出集群定义
rabbitmqadmin export rabbit.definitions.json

# K8s 环境执行
kubectl exec -it rabbitmq-server-0 -- \
  rabbitmqadmin export /tmp/definitions.json

kubectl cp rabbitmq-server-0:/tmp/definitions.json ./backup/definitions.json

# 导入定义（空集群或重建后）
rabbitmqadmin import rabbit.definitions.json

# K8s 环境
kubectl cp ./backup/definitions.json rabbitmq-server-0:/tmp/definitions.json
kubectl exec rabbitmq-server-0 -- \
  rabbitmqadmin import /tmp/definitions.json

apiVersion: snapshot.storage.k8s.io/v1
kind: VolumeSnapshot
metadata:
  name: rabbitmq-data-snapshot-20240501
spec:
  volumeSnapshotClassName: csi-snapshot-class
  source:
    persistentVolumeClaimName: persistence-rabbitmq-server-0

1. 确认备份数据可用性（定义文件 + PVC 快照）
2. 停止 Producer 连接（避免新消息写入）
3. 从 PVC 快照创建新 PVC
4. 部署新 RabbitMQ Cluster（使用恢复的 PVC）
5. 导入 Definitions
6. 验证集群状态和消息完整性
7. 恢复 Producer / Consumer 连接

spec:
  persistence:
    storageClassName: premium-ssd
    storage: 200Gi

所需存储 = 日均消息量 × 平均消息大小 × 保留天数 × 1.5（冗余系数）

示例：
  100 万条/天 × 2KB × 7 天 × 1.5
  = 2GB/天 × 7 × 1.5
  ≈ 21GB

// 连接池 + Channel 池 示例
type RabbitPool struct {
    connPool   []*amqp.Connection
    chPool     chan *amqp.Channel   // Channel 对象池
    poolSize   int
}

func (p *RabbitPool) GetChannel() (*amqp.Channel, error) {
    select {
    case ch := <-p.chPool:
        return ch, nil
    default:
        // 池中无空闲 Channel，创建新 Channel
        conn := p.connPool[rand.Intn(len(p.connPool))]
        return conn.Channel()
    }
}

func (p *RabbitPool) ReturnChannel(ch *amqp.Channel) {
    select {
    case p.chPool <- ch:
    default:
        ch.Close()  // 池满则关闭
    }
}

spec:
  override:
    rabbitmq:
      spec:
        containers:
        - name: rabbitmq
          securityContext:
            capabilities:
              add: ["SYS_RESOURCE"]
          resources:
            limits:
              cpu: "4"
              memory: 8Gi

# rabbitmq.conf
channel_max = 0                     # 每 Connection 最大 Channel 数（0 = 不限制）
heartbeat = 60                      # 心跳间隔（秒）
initial_frame_max = 131072          # 帧最大大小（128KB）

# 确认是否启用 HiPE
rabbitmqctl status | grep hipe

# Docker 镜像启用（使用 3.12+ 默认已启用）
# RabbitMQ 3.12+ 已移除 HiPE，改为 JIT 编译器（性能更优）

# /etc/sysctl.conf
fs.file-max = 655360
net.core.somaxconn = 4096
net.ipv4.tcp_max_syn_backlog = 4096
net.core.rmem_default = 262144
net.core.wmem_default = 262144

# /etc/security/limits.conf
rabbitmq soft nofile 65536
rabbitmq hard nofile 65536

Producer                          Broker
   │                                │
   │──── BasicPublish ──────────────│
   │                                │ 1. 消息入队/持久化
   │                                │ 2. 路由到目标 Queue
   │←─── BasicAck (single) ────────│    成功
   │←─── BasicAck (multiple) ──────│    批量成功
   │←─── BasicNack ────────────────│    失败

// Go 发布确认实现
confirms := ch.NotifyPublish(make(chan amqp.Confirmation, 100))
ch.Confirm(false)  // noWait=false

err := ch.PublishWithContext(ctx, "exchange", "routing.key", false, false, msg)
if err != nil {
    // 连接级错误
}

select {
case confirm := <-confirms:
    if confirm.Ack {
        // 消息已确认
    } else {
        // 消息未确认，需要重试
    }
case <-ctx.Done():
    // 超时
}

// Manual ACK 模式 - 推荐生产使用
msgs, _ := ch.Consume("queue", "consumer-tag", false, false, false, false, nil)

for msg := range msgs {
    // 处理消息
    if process(msg.Body) == nil {
        msg.Ack(false)    // 单条确认
    } else {
        msg.Nack(false, false)  // 拒绝，不 requeue（进入 DLQ）
    }
}

// 异步确认 + 重试簿（Outstanding Confirms Map）
type PublishState struct {
    pending map[uint64]*PendingMsg
    mu      sync.Mutex
}

func (ps *PublishState) HandleConfirm(conf amqp.Confirmation) {
    ps.mu.Lock()
    defer ps.mu.Unlock()
    if msg, ok := ps.pending[conf.DeliveryTag]; ok {
        if conf.Ack {
            delete(ps.pending, conf.DeliveryTag)
        } else {
            // 重试发布
            go ps.retry(msg)
        }
    }
}

内存使用率
100% ┤
     │                              ████████
 80% ┤                        ██████
     │                  ██████
 60% ┤  ← vm_memory_high_watermark (触发流控)
     │            ████
 40% ┤      ██████
     │██████                                  ████
 20% ┤                                          
     │                                          
  0% ┼──────────────────────────────────────────
            正常操作         流控阻断

# rabbitmq.conf
vm_memory_high_watermark.relative = 0.6
vm_memory_calculation_strategy = rss               # 按 RSS 计算内存使用
disk_free_limit.relative = 2.0                     # 2.0 倍相对值
disk_free_limit.absolute = 2GB                     # 或绝对值 2GB

# PrometheusRule 示例
groups:
- name: rabbitmq
  rules:
  - alert: RabbitMQMemoryHigh
    expr: rabbitmq_process_resident_memory_bytes / rabbitmq_resident_memory_limit_bytes > 0.8
    for: 5m
    labels:
      severity: warning
    annotations:
      summary: "RabbitMQ 内存使用率超过 80%"

  - alert: RabbitMQDiskLow
    expr: rabbitmq_disk_space_available_bytes < 5 * 1024^3
    for: 5m
    labels:
      severity: critical
    annotations:
      summary: "RabbitMQ 磁盘剩余空间低于 5GB"

1. kubectl describe pod <pod-name>
   → 检查 Events：PVC 挂载失败？资源不足？ImagePullBackOff？

2. kubectl logs <pod-name>
   → 检查 Erlang Crash Dump
   → 检查 Mnesia 数据库加载错误
   → 检查 Cookie 不匹配错误

3. 进入容器调试
   kubectl exec -it <pod-name> -- bash
   → rabbitmq-diagnostics status
   → rabbitmq-diagnostics check_port_connectivity

# 生产级 RabbitmqCluster CR
apiVersion: rabbitmq.com/v1beta1
kind: RabbitmqCluster
metadata:
  name: prod-rabbitmq
  namespace: messaging
spec:
  replicas: 3
  image: rabbitmq:3.12-management-alpine
  service:
    type: ClusterIP
    annotations:
      prometheus.io/scrape: "true"
      prometheus.io/port: "15692"
  persistence:
    storageClassName: premium-ssd
    storage: 200Gi
  resources:
    requests:
      cpu: "2"
      memory: 4Gi
    limits:
      cpu: "4"
      memory: 8Gi
  affinity:
    podAntiAffinity:
      requiredDuringSchedulingIgnoredDuringExecution:
      - labelSelector:
          matchLabels:
            app.kubernetes.io/name: rabbitmq
        topologyKey: kubernetes.io/hostname
  tolerations: []
  rabbitmq:
    additionalConfig: |
      vm_memory_high_watermark.relative = 0.6
      disk_free_limit.relative = 2.0
      cluster_partition_handling = pause_minority
      cluster_keepalive_interval = 10000
      collect_statistics_interval = 30000
      log.console.level = info
      queue_master_locator = client-local      
    additionalPlugins:
    - rabbitmq_prometheus
    - rabbitmq_delayed_message_exchange
  tls:
    secretName: rabbitmq-tls
    disableNonTLSListeners: true

节点数规划（Quorum Queue）：
  - 3 节点：多数派可容忍 1 节点故障
  - 5 节点：多数派可容忍 2 节点故障

PVC 容量规划：
  日均消息数 × 平均大小 × 保留天数 × 安全系数(1.5~2.0)

CPU 规划：
  每核约处理 20K-50K msg/s（取决于消息大小和队列类型）

垂直扩展 vs 水平扩展：
  - 优先垂直扩展（增加 CPU/Memory Limit）
  - 水平扩展（增加节点数）用于提升 Quorum Queue 容错