消息队列可观测性：Kafka 与 RabbitMQ 的滞后、积压与端到端延迟

生产速率（Produce Rate）：每秒进入多少条
消费速率（Consume Rate）：每秒处理多少条
积压量（Lag / Depth）：还有多少条没被处理
端到端延迟（E2E Latency）：一条消息从生产到被消费经历多久

关系：积压变化率 = 生产速率 - 消费速率
关键：积压量大不等于有问题，积压"持续增长"才是问题

场景 A：大促开始，生产速率从 1k/s 涨到 10k/s，消费 10k/s
        → 积压不增长，但系统已在高负载，需要预警
场景 B：下游某依赖变慢，消费从 10k/s 掉到 2k/s
        → 积压快速累积，必须立刻告警
场景 C：消息体变大，单条处理时间变长
        → 消费速率不变但"延迟"变长，积压指标看不出来

结论：必须同时看速率、积压与延迟，单一指标必然误导

自上而下：先确认 Broker 健康，再看队列积压，最后定位消费者

Lag = LogEndOffset(LEO) - CommittedOffset

LEO：分区日志末尾的偏移量（已写入的消息总数）
CommittedOffset：消费者组已提交确认的偏移量
Lag：还有多少条未确认消费

注意：lag 是"分区级"的，topic 级 lag = 所有分区之和

Broker 侧：
  kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec
  kafka.server:type=ReplicaManager,name=UnderReplicatedPartitions
  kafka.server:type=ReplicaManager,name=IsrShrinksPerSec
  kafka.log:type=LogFlushStats,name=LogFlushRateAndTimeMs

Consumer 侧：
  kafka.consumer:type=consumer-fetch-manager-metrics,client-id=*
    records-lag-max          最大滞后
    records-consumed-rate    消费速率
    fetch-latency-avg        拉取延迟
  kafka.consumer:type=consumer-coordinator-metrics
    rebalance-rate-per-hour  再平衡频率

部署 kafka-exporter，暴露 Prometheus 指标：
  kafka_consumergroup_lag{consumergroup,topic,partition}
  kafka_consumergroup_current_offset
  kafka_topic_partition_current_offset
  kafka_topic_partitions
  kafka_brokers

告警表达式（PromQL）：
  # 单分区滞后超过 10 万
  kafka_consumergroup_lag > 100000

  # 滞后持续增长（5 分钟斜率 > 0 且超过阈值）
  deriv(kafka_consumergroup_lag[5m]) > 0 and kafka_consumergroup_lag > 10000

  # 消费速率骤降
  rate(kafka_consumergroup_current_offset[5m]) < 0.2 * rate(kafka_consumergroup_current_offset[1h] offset 1d)

问题：某 topic 有 24 个分区，但 lag 集中在 3 个分区
根因：
  - 分区键（key）分布不均 → 热点 key
  - 消费者分配不均 → 部分实例拿到更多分区
对策：
  - 监控 max(lag) by (partition)，而不是只看 sum
  - 热点 key 加盐或改分区策略
  - 检查消费者分区分配策略（Sticky/Range/CooperativeSticky）

rabbitmq_queue_messages            队列中总消息数（含未确认）
rabbitmq_queue_messages_ready      可投递消息数（真正积压）
rabbitmq_queue_messages_unacknowledged  已投递未确认数
rabbitmq_queue_consumers           消费者数量
rabbitmq_queue_messages_published_total  发布总数
rabbitmq_queue_messages_delivered_total  投递总数
rabbitmq_queue_messages_ack_total        确认总数

关键区分：
  messages_ready 高 → 没有消费者或消费者太慢（真积压）
  unacknowledged 高 → 消费者拿到但不 ack（处理慢或卡死）

告警建议：
  messages_ready > 阈值 且持续 5 分钟
  consumers == 0（队列无人消费，最危险）
  unacknowledged 持续增长（消费者疑似 hang）

PromQL：
  # 每秒确认速率
  rate(rabbitmq_queue_messages_ack_total[5m])

  # 每秒发布速率
  rate(rabbitmq_queue_messages_published_total[5m])

  # 积压增长速率
  deriv(rabbitmq_queue_messages_ready[10m])

1. 流控（Flow Control）
   触发条件：内存或磁盘水位超阈值
   现象：broker 阻塞生产者，发布速率骤降
   指标：rabbitmq_connections_state、内存告警

2. 消费者预取（prefetch）配置不当
   prefetch 过大 → 单消费者囤积消息，其他消费者空闲
   prefetch 过小 → 网络往返多，吞吐上不去
   建议：prefetch 从 30~100 起调，按处理耗时与实例数权衡

积压 = 0 不代表延迟低：
  一条消息 10:00 生产、10:00:01 消费 → 延迟 1s，积压为 0
  一条消息 09:00 生产、10:00 消费 → 延迟 1h，但此刻积压也是 0（已被消费）

结论：延迟是"已消费消息的历史"，积压是"当前快照"，必须分别度量

方案一：消息头携带时间戳
  生产端：在 header 写入 produce_ts（或 trace 上下文）
  消费端：now - produce_ts → 端到端延迟
  注意：跨服务时钟必须同步（NTP/PTP）

方案二：OpenTelemetry 消息语义约定
  生产：创建 PRODUCER span，注入 traceparent 到消息头
  消费：提取上下文，创建 CONSUMER span 作为子 span
  属性：messaging.system、messaging.destination.name、
        messaging.kafka.partition、messaging.message.id
  效果：链路里直接看到"消息在队列里等了多久"（span 之间的空隙）

指标：messaging_e2e_latency_seconds（Histogram）
  label：topic / queue、consumer_group

PromQL：
  histogram_quantile(0.99,
    sum by (le, topic) (rate(messaging_e2e_latency_seconds_bucket[5m]))) > 60

含义：P99 端到端延迟超过 60 秒 → 说明消息在系统里滞留过久

乱序：
  Kafka 单分区内有序，跨分区无序 → 需要按业务键分区
  观测：比较消费顺序与生产顺序（可用序列号）
重复：
  至少一次语义下重复不可避免 → 消费端需幂等
  观测：统计 message_id 去重命中率

再平衡（Rebalance）= 消费者组重新分配分区
触发：消费者加入/离开、订阅变化、心跳超时、处理超时
代价：
  再平衡期间全组停止消费（Eager 协议）
  分区越多、成员越多，再平衡越慢
  频繁再平衡 = 周期性消费停顿 → 积压呈锯齿状

kafka.consumer:type=consumer-coordinator-metrics
  rebalance-rate-per-hour        再平衡频率（> 2 次/小时要查）
  rebalance-latency-avg          再平衡平均耗时
  assigned-partitions            分配到的分区数
  last-rebalance-seconds-ago     距上次再平衡时间
  heartbeat-rate / heartbeat-response-time-max

诊断：
  再平衡频繁 + 心跳超时 → 消费者处理时间超过 max.poll.interval.ms
  对策：调大 max.poll.interval.ms 或减小 max.poll.records
  再平衡频繁 + 实例频繁重启 → 检查 OOM / 探针失败

Eager（默认旧协议）：全部撤销 → 重新分配，全组停顿
CooperativeSticky（推荐）：只迁移必要分区，增量式，停顿小
配置：partition.assignment.strategy=CooperativeStickyAssignor
收益：再平衡停顿从"秒级"降到"毫秒级"

P0：consumers == 0（队列无人消费）或 lag 增长速率极高
P1：lag 超过阈值且持续增长，或再平衡频率异常
P2：消费速率下降但未积压，或分区倾斜
P3：broker 磁盘/内存水位预警

反模式：
  lag > 1000 就告警 → 正常波动就触发，噪音大
正确做法：
  用 for: 10m 持续条件
  用增长速率而非绝对值（deriv > 0）
  区分"业务高峰的正常积压"与"异常积压"（按时间基线）
  维护窗口与批量任务期间静默

消费能力估算：
  单消费者吞吐 = 1 / 单条处理耗时 × 并发数
  所需消费者数 = 峰值生产速率 / 单消费者吞吐 × 安全系数(1.5)

分区数规划：
  分区数 ≥ 消费者实例数（否则有空闲消费者）
  分区数一旦确定难以下调 → 预留 2~3 倍增长空间

磁盘容量：
  积压容忍量 × 单条消息平均大小 × 副本数
  例：容忍 1000 万条积压 × 1KB × 3 副本 ≈ 30GB

□ 同时监控生产速率、消费速率、积压与端到端延迟四类信号
□ 积压告警用 max by (partition)，避免倾斜被平均
□ 用增长速率（deriv）而非绝对值做积压告警
□ 监控 consumers == 0，这是最危险的静默故障
□ 用 OpenTelemetry 消息语义约定做端到端追踪
□ 消息头携带生产时间戳，全链路时钟同步
□ 监控再平衡频率与 max.poll.interval.ms 的关系
□ 采用 CooperativeSticky 减少再平衡停顿
□ 消费端幂等，按 message_id 去重
□ 按峰值速率 × 安全系数规划消费者与分区数

消息队列可观测性 = 速率 + 积压 + 延迟三件套，
按分区看倾斜、按增长看趋势、按链路看滞留。