RabbitMQ 深度实践：在 Kubernetes 上构建生产级消息队列

channel.confirmSelect();
channel.basicPublish(exchange, routingKey, props, body);
if (!channel.waitForConfirms(5000)) {
    // 重试或告警
}

kubectl apply -f "https://github.com/rabbitmq/cluster-operator/releases/latest/download/cluster-operator.yml"

kubectl get pods -n rabbitmq-system
# NAME                                         READY   STATUS    RESTARTS   AGE
# rabbitmq-cluster-operator-7cbf865f84-xnlwz   1/1     Running   0          30s

apiVersion: rabbitmq.com/v1beta1
kind: RabbitmqCluster
metadata:
  name: rabbitmq-production
  namespace: messaging
spec:
  # 三节点：满足 Quorum Queue 的 Raft 多数派要求（2/3 节点存活即可选主）
  replicas: 3
  image: rabbitmq:3.13-management

  resources:
    requests:
      cpu: "1"
      memory: 2Gi
    limits:
      cpu: "2"
      memory: 4Gi

  # 持久化存储：使用高性能 SSD，避免 HDD 在 Raft 日志同步时成为瓶颈
  persistence:
    storageClassName: fast-ssd
    storage: 50Gi

  tls:
    secretName: rabbitmq-tls
    disableNonTLSListeners: true  # 生产环境强制 TLS

  # Pod 反亲和性：确保三个节点分布在不同物理机，避免单点故障
  affinity:
    podAntiAffinity:
      requiredDuringSchedulingIgnoredDuringExecution:
        - labelSelector:
            matchLabels:
              app.kubernetes.io/name: rabbitmq-production
          topologyKey: kubernetes.io/hostname

  # 拓扑分布约束：跨可用区均匀分布（多 AZ 场景）
  topologySpreadConstraints:
    - maxSkew: 1
      topologyKey: topology.kubernetes.io/zone
      whenUnsatisfiable: DoNotSchedule
      labelSelector:
        matchLabels:
          app.kubernetes.io/name: rabbitmq-production

  rabbitmq:
    additionalPlugins:
      - rabbitmq_prometheus
      - rabbitmq_shovel
      - rabbitmq_shovel_management

    additionalConfig: |
      # 网络分区处理：少数派节点暂停，避免脑裂（详见第四章）
      cluster_partition_handling = pause-minority

      # 内存水位：超过 60% 物理内存时触发流控，阻塞生产者
      vm_memory_high_watermark.relative = 0.6

      # 磁盘空间告警：剩余空间低于 1.5 倍内存时告警
      disk_free_limit.relative = 1.5

      # 默认队列类型：新建队列默认使用 Quorum Queue（与第二章选型一致）
      default_queue_type = quorum

      # 消费者超时：60 分钟未 ack 则认为消费者异常
      consumer_timeout = 3600000

apiVersion: cert-manager.io/v1
kind: Certificate
metadata:
  name: rabbitmq-tls
  namespace: messaging
spec:
  secretName: rabbitmq-tls
  duration: 2160h   # 90 天
  renewBefore: 360h # 提前 15 天自动续签
  dnsNames:
    - rabbitmq-production.messaging.svc.cluster.local
    - "*.rabbitmq-production-nodes.messaging.svc.cluster.local"
  issuerRef:
    name: cluster-ca-issuer
    kind: ClusterIssuer

apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: rabbitmq-network-policy
  namespace: messaging
spec:
  podSelector:
    matchLabels:
      app.kubernetes.io/name: rabbitmq-production
  policyTypes:
    - Ingress
    - Egress
  ingress:
    - ports:
        - port: 5671   # AMQP over TLS
          protocol: TCP
    - ports:
        - port: 15671  # 管理 UI（仅允许运维网段）
          protocol: TCP
      from:
        - namespaceSelector:
            matchLabels:
              purpose: ops
    - ports:
        - port: 15692  # Prometheus 指标采集
          protocol: TCP
      from:
        - namespaceSelector:
            matchLabels:
              purpose: monitoring
    - ports:
        - port: 25672  # 集群间通信（Erlang distribution）
          protocol: TCP
      from:
        - podSelector:
            matchLabels:
              app.kubernetes.io/name: rabbitmq-production
  egress:
    - ports:
        - port: 25672
          protocol: TCP
        - port: 4369
          protocol: TCP  # epmd（Erlang port mapper daemon）
    - ports:
        - port: 53
          protocol: UDP  # DNS 解析

# 升级前确认 Quorum 状态正常
kubectl exec -it rabbitmq-production-server-0 -n messaging -- \
  rabbitmq-diagnostics quorum_status

# 确认无消息积压
kubectl exec -it rabbitmq-production-server-0 -n messaging -- \
  rabbitmqctl list_queues name messages consumers

# 检查分区状态
rabbitmqctl cluster_status | grep partitions

# 修复网络连通性后，少数派节点自动重新加入
# 如果节点未自动恢复，手动重启节点
kubectl rollout restart statefulset/rabbitmq-production-server -n messaging

# 验证恢复
rabbitmq-diagnostics cluster_status

// prefetch=0：无限制，消费者一次性拉取所有消息（危险！）
// prefetch=1：严格公平分发，吞吐量低但保证负载均衡
// prefetch=100：推荐平衡值，根据消息处理时间调整
channel.basicQos(100);

apiVersion: monitoring.coreos.com/v1
kind: ServiceMonitor
metadata:
  name: rabbitmq-production
  namespace: messaging
  labels:
    release: prometheus
spec:
  selector:
    matchLabels:
      app.kubernetes.io/name: rabbitmq-production
      app.kubernetes.io/component: rabbitmq
  endpoints:
    - port: prometheus
      scheme: https
      tlsConfig:
        caFile: /etc/prometheus/secrets/rabbitmq-tls/ca.crt
        certFile: /etc/prometheus/secrets/rabbitmq-tls/tls.crt
        keyFile: /etc/prometheus/secrets/rabbitmq-tls/tls.key
        insecureSkipVerify: false
      interval: 15s
      scrapeTimeout: 10s
  namespaceSelector:
    matchNames:
      - messaging

# 告警：任意队列超过 10000 条未消费消息，持续 5 分钟
sum by (queue, vhost) (rabbitmq_queue_messages{state="ready"}) > 10000

# 告警：unacked 持续增长超过 15 分钟（消费者处理能力不足或卡死）
deriv(rabbitmq_queue_messages_unacked[5m]) > 0

# 告警：节点内存使用超过 limit 的 80%（接近流控水位）
rabbitmq_process_memory_bytes / rabbitmq_resident_memory_limit_bytes > 0.8

# 告警：1 分钟内新增连接数超过 200（可能是客户端重连风暴）
increase(rabbitmq_connections_opened_total[1m]) > 200

groups:
  - name: rabbitmq.rules
    rules:
      - alert: RabbitMQQueueDepthHigh
        expr: |
          sum by (queue, vhost, namespace) (
            rabbitmq_queue_messages{state="ready"}
          ) > 10000
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "RabbitMQ 队列积压过高"
          description: "队列 {{ $labels.queue }} 积压 {{ $value }} 条消息，请检查消费者状态"

      - alert: RabbitMQNodeDown
        expr: up{job="rabbitmq"} == 0
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "RabbitMQ 节点不可达"
          description: "节点 {{ $labels.instance }} 已无法采集指标超过 1 分钟"

      - alert: RabbitMQHighMemoryUsage
        expr: |
          rabbitmq_process_memory_bytes
          / rabbitmq_resident_memory_limit_bytes > 0.8
        for: 2m
        labels:
          severity: warning
        annotations:
          summary: "RabbitMQ 内存使用率过高"
          description: "节点 {{ $labels.instance }} 内存使用率 {{ $value | humanizePercentage }}，接近流控水位"

      - alert: RabbitMQQuorumQueueMinority
        expr: |
          rabbitmq_quorum_queue_disallow_changes == 1
        for: 30s
        labels:
          severity: critical
        annotations:
          summary: "RabbitMQ Quorum Queue 进入少数派状态"
          description: "集群节点数不足以维持 Quorum，队列已停止接受写入"

      - alert: RabbitMQLowDiskSpace
        expr: |
          rabbitmq_disk_space_available_bytes
          / rabbitmq_disk_space_available_limit_bytes < 1.5
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "RabbitMQ 磁盘空间告警"
          description: "节点 {{ $labels.instance }} 可用磁盘空间接近告警阈值"

      - alert: RabbitMQCertExpiringSoon
        expr: |
          (rabbitmq_tls_certificates_expiry_seconds - time()) / 86400 < 30
        labels:
          severity: warning
        annotations:
          description: "RabbitMQ TLS 证书将在 {{ $value | humanizeDuration }} 后过期"

log.console = true
log.console.level = info
log.console.formatter = json

pipelineStages:
  - json:
      expressions:
        level: level
        msg: msg
        node: node
  - labels:
      level:
      node:

// 1. Quorum Queue（强制持久，替代需手动设置 durable 的 Classic Queue）
channel.queueDeclare("work-queue", true, false, false, Map.of("x-queue-type", "quorum"));

// 2. 持久化消息（deliveryMode=2，消息落盘，Broker 重启后不丢失）
AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
    .deliveryMode(2)
    .build();
channel.basicPublish(exchange, routingKey, props, body);

// 3. Publisher Confirms（确认消息已被 Broker 多数派持久化）
channel.confirmSelect();
channel.basicPublish(exchange, routingKey, props, body);
channel.waitForConfirmsOrDie(5000);

// 方案1：基于消息 ID 去重（Redis SET NX）
String msgId = properties.getMessageId();
if (redis.set(msgId, "1", SetArgs.Builder.nx().ex(86400)) == null) {
    channel.basicAck(deliveryTag, false);  // 重复消息，直接 ack
    return;
}
// 业务处理...
channel.basicAck(deliveryTag, false);

// 方案2：数据库唯一约束
// INSERT INTO orders (id, ...) VALUES (?, ...) ON CONFLICT (id) DO NOTHING;

# 立即临时提高水位（应急操作）
rabbitmqctl set_vm_memory_high_watermark 0.7

# 排查积压队列（内存消耗主因）
rabbitmqctl list_queues name messages memory --sorted-by messages

# 扩容消费者处理积压
kubectl scale deployment my-consumer --replicas=10

# 检查是否有死连接持有大量 unacked 消息
rabbitmqctl list_connections name state channels

# 确认分区情况
rabbitmqctl cluster_status

# 确定"权威"节点（通常是消息数更多的节点）
rabbitmqctl list_queues name messages --node rabbit@node1
rabbitmqctl list_queues name messages --node rabbit@node2

# 重启少数派节点（数据丢弃并从多数派同步）
kubectl delete pod rabbitmq-production-server-1 -n messaging

# 验证集群恢复
rabbitmqctl cluster_status | grep partitions  # 应返回空

# 检查消费者 prefetch 设置
rabbitmqctl list_consumers queue_name prefetch_count ack_required

# 检查 unacked 消息分布
rabbitmqctl list_queues name messages_unacked messages_ready

// 修改消费者代码，设置合理 prefetch（参考第四章调优原则）
channel.basicQos(100);

# 查看消息 publish/consume 速率
rabbitmqctl list_queues name messages publish_rate deliver_rate

# 扩容消费者
kubectl scale deployment slow-consumer --replicas=20

# 为队列设置上限，防止无限积压导致 OOM
rabbitmqctl set_policy ha-all "^" \
  '{"max-length":100000,"overflow":"reject-publish","dead-letter-exchange":"dlx"}' \
  --apply-to queues

// 错误示例：固定间隔重连
Thread.sleep(1000);
connection.reconnect();

// 正确示例：指数退避 + 抖动
int attempt = 0;
while (!connected) {
    long delay = Math.min(30000, (long)(1000 * Math.pow(2, attempt)));
    long jitter = (long)(Math.random() * delay * 0.3);  // 30% 随机抖动
    Thread.sleep(delay + jitter);
    try {
        connection = factory.newConnection();
        connected = true;
    } catch (Exception e) {
        attempt++;
    }
}

spring:
  rabbitmq:
    listener:
      simple:
        retry:
          enabled: true
          initial-interval: 1000ms
          max-interval: 30000ms
          multiplier: 2.0

# 检查1：是否使用了 Quorum Queue 或 durable Classic Queue
rabbitmqctl list_queues name durable

# 检查2：消息是否设置 deliveryMode=2（在管理 UI 手动发布测试消息验证）

# 检查3：生产者是否开启了 Publisher Confirms
# 未开启时，Broker 异常可能在 ack 返回前丢失消息

# 检查4：是否有 DLX 配置导致消息被路由到死信队列
rabbitmqctl list_queues name dead_letter_exchange messages

# 检查5：队列是否设置了 TTL 导致消息过期
rabbitmqctl list_queues name message_ttl

# 监控证书续签事件
kubectl get events -n messaging --field-selector reason=CertIssued

# 验证新证书已生效
openssl s_client -connect rabbitmq-production.messaging.svc:5671 \
  -CAfile /path/to/ca.crt 2>/dev/null | openssl x509 -noout -dates