消息队列监控实战：Kafka Lag、RabbitMQ 积压与消费延迟告警

Lag = LogEndOffset - CurrentOffset

# docker-compose.yml
services:
  kafka-exporter:
    image: danielqsj/kafka_exporter:v1.9.0
    command:
      - --kafka.server=kafka1:9092
      - --kafka.server=kafka2:9092
      - --kafka.server=kafka3:9092
      - --topic.filter=^.*$          # 采集所有 topic
      - --group.filter=^.*$          # 采集所有消费组
      - --concurrent.enable          # 并发采集，大集群必开
      - --concurrent.max=200
      - --log.enable-sarama=false   # 关掉 sarama 日志刷屏
    ports:
      - "9308:9308"
    restart: unless-stopped

# 当前消费组已提交 offset
kafka_consumergroup_current_offset{consumergroup="order-consumer",topic="orders",partition="0"} 12345678
# 分区最新 offset（LEO）
kafka_topic_partition_leader{topic="orders",partition="0"} 12355678
# 算出来的 Lag（exporter 自动算好）
kafka_consumergroup_lag{consumergroup="order-consumer",topic="orders",partition="0"} 10000

# kafka 启动参数加一行
KAFKA_OPTS="$KAFKA_OPTS -javaagent:/opt/jmx_exporter/jmx_prometheus_javaagent-0.20.0.jar=9404:/opt/jmx_exporter/kafka.yml"

# 告警规则文件：kafka-lag-alerts.yml
groups:
  - name: kafka_consumer_lag
    interval: 30s
    rules:
      # P2：单分区 Lag 超过 5000，持续 5 分钟
      - alert: KafkaConsumerLagHigh
        expr: |
          sum by (consumergroup, topic) (
            kafka_consumergroup_lag
          ) > 5000          
        for: 5m
        labels:
          severity: P2
        annotations:
          summary: "Kafka 消费组 {{ $labels.consumergroup }} 的 {{ $labels.topic }} Lag 过高"
          description: "当前总 Lag: {{ $value }}，超过阈值 5000"

      # P1：Lag 持续增长 10 分钟未回落
      - alert: KafkaConsumerLagIncreasing
        expr: |
          deriv(
            sum by (consumergroup, topic) (
              kafka_consumergroup_lag
            )[10m:1m]
          ) > 0          
        for: 10m
        labels:
          severity: P1
        annotations:
          summary: "Kafka Lag 持续增长: {{ $labels.consumergroup }} / {{ $labels.topic }}"
          description: "Lag 在过去 10 分钟内持续上升，消费者可能已停止或处理能力不足"

      # P0：Lag 突然归零（疑似消费者挂掉后停止提交）
      - alert: KafkaConsumerLagDroppedToZero
        expr: |
          (sum by (consumergroup, topic) (kafka_consumergroup_lag)) == 0
          and
          (sum by (consumergroup, topic) (kafka_consumergroup_lag offset 5m)) > 10000          
        for: 2m
        labels:
          severity: P0
        annotations:
          summary: "Kafka Lag 突然归零: {{ $labels.consumergroup }}"
          description: "5 分钟前 Lag > 10000，现在归零，疑似消费者崩溃或 offset 异常"

# 在每个 RabbitMQ 节点执行
rabbitmq-plugins enable rabbitmq_prometheus
# 默认暴露在 15692 端口

groups:
  - name: rabbitmq_queue_depth
    interval: 30s
    rules:
      # P1：消费者全部消失
      - alert: RabbitMQNoConsumers
        expr: rabbitmq_consumers == 0
        for: 1m
        labels:
          severity: P1
        annotations:
          summary: "RabbitMQ 队列 {{ $labels.queue }} 没有消费者"
          description: "队列 {{ $labels.queue }} 在 {{ $labels.vhost }} 下无消费者，消息会一直堆积"

      # P2：队列深度持续过高
      - alert: RabbitMQQueueBacklog
        expr: rabbitmq_queue_messages > 10000
        for: 5m
        labels:
          severity: P2
        annotations:
          summary: "RabbitMQ 队列 {{ $labels.queue }} 积压"
          description: "队列 {{ $labels.queue }} 消息数: {{ $value }}"

      # P1：unacked 消息过多（消费者处理慢或卡住）
      - alert: RabbitMQUnackedHigh
        expr: rabbitmq_queue_messages_unacknowledged > 5000
        for: 3m
        labels:
          severity: P1
        annotations:
          summary: "RabbitMQ unacked 消息过多"
          description: "队列 {{ $labels.queue }} 有 {{ $value }} 条 unacked 消息，消费者可能卡住"

      # P3：死信队列有消息（业务异常，不紧急但需关注）
      - alert: RabbitMQDeadLetterQueue
        expr: |
          rabbitmq_queue_messages{queue=~".*\\.dlq$|.*dead_letter.*"} > 0          
        for: 10m
        labels:
          severity: P3
        annotations:
          summary: "死信队列 {{ $labels.queue }} 有消息"
          description: "死信队列有 {{ $value }} 条消息，检查上游消费失败原因"

# prometheus-redis-exporter
services:
  redis-exporter:
    image: oliver006/redis_exporter:v1.66.0
    command:
      - --redis.addr=redis:6379
      - --redis.password=${REDIS_PASSWORD}
      # 检查指定 key 的 List 长度
      - --check-key-groups=true
      - --check-keys=queue:*,task:*,*:pending
    ports:
      - "9121:9121"

redis_key_size{key="queue:order_tasks"} 1532
redis_key_size{key="queue:email_tasks"} 0

groups:
  - name: redis_queue_depth
    interval: 30s
    rules:
      - alert: RedisQueueBacklog
        expr: redis_key_size{key=~"queue:.*"} > 5000
        for: 5m
        labels:
          severity: P2
        annotations:
          summary: "Redis 队列 {{ $labels.key }} 积压"
          description: "队列长度: {{ $value }}"

      # 延迟队列（Sorted Set）的超时检测
      - alert: RedisDelayedTaskOverdue
        expr: |
          redis_db_keys{db="0"}  # 结合自定义脚本
          # 更好的做法：写个脚本定期 ZCOUNT 统计到期未处理的任务数          
        for: 5m
        labels:
          severity: P2
        annotations:
          summary: "Redis 延迟队列有过期未处理任务"

#!/usr/bin/env python3
"""redis_delayed_queue_exporter.py
定期扫描 Redis Sorted Set 类型的延迟队列，
统计已到期但尚未被消费的任务数量。
"""
import time
import redis
from prometheus_client import CollectorRegistry, Gauge, write_to_textfile

r = redis.Redis(host='redis', port=6379, decode_responses=True)
registry = CollectorRegistry()

overdue_tasks = Gauge(
    'redis_delayed_queue_overdue',
    'Number of overdue tasks in delayed queue',
    ['queue_name'],
    registry=registry
)

now = time.time()
# 扫描所有延迟队列 key（约定 key 前缀为 delayed:）
for key in r.scan_iter(match='delayed:*'):
    # ZCOUNT key min max: 统计 score <= now 的元素数
    count = r.zcount(key, 0, now)
    overdue_tasks.labels(queue_name=key).set(count)

write_to_textfile('/var/lib/node_exporter/textfile/redis_delayed.prom', registry)

# 1. 查看消费组状态和 Lag
kafka-consumer-groups.sh --bootstrap-server kafka1:9092 \
  --describe --group order-consumer

# 输出示例：
# GROUP           TOPIC       PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG     CONSUMER-ID
# order-consumer  orders      0          12345678        12355678        10000   consumer-1
# order-consumer  orders      1          23456789        23456789        0       consumer-1
# order-consumer  orders      2          34567890        34567990        100     consumer-2
# order-consumer  orders      3          -               45678901        45678901 -  ← 没有消费者！

# 2. 查看消费组成员（确认消费者是否活着）
kafka-consumer-groups.sh --bootstrap-server kafka1:9092 \
  --describe --group order-consumer --members --verbose

# 3. 查看 topic 的分区分布
kafka-topics.sh --bootstrap-server kafka1:9092 \
  --describe --topic orders

# 4. 实时监控 Lag 变化（每 5 秒刷新）
watch -n 5 'kafka-consumer-groups.sh --bootstrap-server kafka1:9092 --describe --group order-consumer | tail -n +3 | sort -k6 -rn'

# 查看队列详情（消费者数、消息数、unacked 数）
rabbitmqctl list_queues name messages consumers messages_unacknowledged

# 查看消费者连接
rabbitmqctl list_consumers | grep <queue_name>

# 查看连接详情（排查消费者卡住）
rabbitmqctl list_connections name peer_host state channels

# ScaledObject: 根据 Kafka Lag 自动扩缩容消费者
apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: kafka-order-consumer
  namespace: production
spec:
  scaleTargetRef:
    name: order-consumer-deployment
  minReplicaCount: 2        # 最少 2 个副本
  maxReplicaCount: 20       # 最多 20 个副本
  pollingInterval: 30       # 每 30 秒检查一次 Lag
  cooldownPeriod: 300        # 缩容冷却 5 分钟，防抖
  triggers:
    - type: kafka
      metadata:
        bootstrapServers: kafka1:9092,kafka2:9092,kafka3:9092
        consumerGroup: order-consumer
        topic: orders
        lagThreshold: "1000"   # 单分区 Lag 超过 1000 就扩容
        offsetResetPolicy: latest
        partitionLimitation: "0,1,2,3"  # 可选：只监控指定分区

# application.yml
spring:
  kafka:
    consumer:
      group-id: order-consumer
      max-poll-records: 100          # 每次最多拉 100 条
      max-poll-interval-ms: 300000   # 两次 poll 间隔上限 5 分钟
      properties:
        max.partition.fetch.bytes: 1048576  # 每分区单次拉取上限 1MB

# 按 group 汇总总 Lag，降序排列
topk(20, sum by (consumergroup) (kafka_consumergroup_lag))

# Lag 趋势（时间序列图）
sum by (topic) (kafka_consumergroup_lag{consumergroup="$group"})

# 各分区 Lag（表格，高亮最大值）
kafka_consumergroup_lag{consumergroup="$group", topic="$topic"}

# 消费速率（每秒消费的消息数）
rate(kafka_consumergroup_current_offset{consumergroup="$group"}[1m])

# 生产速率（每秒新增的消息数）
rate(kafka_topic_partition_leader{topic="$topic"}[1m])