152｜Kafka 消息积压治理：发现、扩容、并行度、限流和应急消费

积压量(Lag) = Log End Offset(LEO) - Consumer Offset

# 查看所有消费者组的 Lag 概况
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list

# 查看特定消费者组的详细 Lag（含每个分区）
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group order-consumer-group --describe

# 输出示例：
# TOPIC          PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
# order-topic    0          1523400         1525890         2490
# order-topic    1          1521000         1525890         4890  ← 分区1积压严重
# order-topic    2          1525800         1525890         90

kafka.consumer:type=consumer-fetch-manager-metrics,client-id=*
  → records-lag-max  (最大积压)
  → records-lag-avg  (平均积压)

long produceTime = record.timestamp();  // Kafka 消息自带的 timestamp
long consumeTime = System.currentTimeMillis();
metricsCollector.recordLag(consumeTime - produceTime);

# 查看每个分区的 Lag 分布
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group my-group --describe | awk '{if(NR>1) print $3, $6}' | sort -k2 -rn

最大有效消费者数 = 分区数

积压发生
    │
    ├─ 分区数 > 当前消费者数？
    │   ├─ 是 → 增加消费者实例（最快速）
    │   └─ 否 → 继续
    │
    ├─ 消费逻辑有优化空间？
    │   ├─ 是 → 优化消费逻辑（批量处理、异步化、去重）
    │   └─ 否 → 继续
    │
    ├─ 可以增加分区数？
    │   ├─ 是 → 增加分区 + 增加消费者（注意：有 Key 的消息会重新分配）
    │   └─ 否 → 继续
    │
    └─ 紧急 → 启动应急消费程序（见第 7 节）

# 增加分区（从 8 → 16）
kafka-topics.sh --bootstrap-server localhost:9092 \
  --alter --topic order-topic --partitions 16

// 消费者配置要点
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "order-consumer-group");
props.put("enable.auto.commit", "false");        // 手动提交，避免重复消费
props.put("max.poll.records", "500");             // 单次拉取记录数，根据处理能力调优
props.put("max.poll.interval.ms", "300000");      // 两次 poll 最大间隔，超时触发 Rebalance
props.put("session.timeout.ms", "30000");          // 心跳超时
props.put("heartbeat.interval.ms", "10000");       // 心跳间隔，应小于 session.timeout.ms 的 1/3

TPS = 1 / 平均处理时间(秒)

ExecutorService executor = Executors.newFixedThreadPool(20);

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
    
    List<Future<?>> futures = new ArrayList<>();
    for (ConsumerRecord<String, String> record : records) {
        futures.add(executor.submit(() -> {
            processMessage(record);  // 业务处理
        }));
    }
    
    // 等待所有任务完成后再提交 Offset
    for (Future<?> future : futures) {
        future.get();  // 阻塞等待
    }
    consumer.commitSync();  // 全部成功后才提交
}

// 手动 assign 分区，每个分区一个处理线程
consumer.assign(Arrays.asList(
    new TopicPartition("order-topic", 0),
    new TopicPartition("order-topic", 1)
));

// 每个分区使用独立线程处理
for (TopicPartition tp : consumer.assignment()) {
    executor.submit(() -> {
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
            for (ConsumerRecord<String, String> record : records.records(tp)) {
                processMessage(record);
            }
        }
    });
}

List<ConsumerRecord<String, String>> buffer = new ArrayList<>();

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
    for (ConsumerRecord<String, String> record : records) {
        buffer.add(record);
        if (buffer.size() >= 200) {
            batchInsertToDB(buffer);  // 批量写入
            buffer.clear();
            consumer.commitSync();
        }
    }
}

线程数 = CPU核数 × (1 + 平均等待时间 / 平均计算时间)

RateLimiter rateLimiter = RateLimiter.create(500.0);  // 每秒 500 条

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
    for (ConsumerRecord<String, String> record : records) {
        rateLimiter.acquire();  // 阻塞直到获取到令牌
        processMessage(record);
    }
    consumer.commitSync();
}

// Sentinel 流控规则
FlowRule rule = new FlowRule();
rule.setResource("kafka-consume");
rule.setGrade(RuleConstant.FLOW_GRADE_QPS);
rule.setCount(500);  // QPS 上限

FlowRuleManager.loadRules(Collections.singletonList(rule));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
    for (ConsumerRecord<String, String> record : records) {
        Entry entry = null;
        try {
            entry = SphU.entry("kafka-consume");
            processMessage(record);
        } catch (BlockException e) {
            // 被限流，可以选择等待后重试
            Thread.sleep(10);
        } finally {
            if (entry != null) entry.exit();
        }
    }
}

// 伪代码
while (true) {
    double dbLoad = getDatabaseLoad();  // 查询数据库 CPU 使用率、连接数等
    double rate = calculateRate(dbLoad);  // 根据负载计算合适的消费速率
    
    rateLimiter.setRate(rate);  // 动态调整限流速率
    
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
    // ... 限流消费
}

boolean emergencyMode = isEmergencyMode();  // 从配置中心读取

void processMessage(ConsumerRecord<String, String> record) {
    // 核心逻辑：必须执行
    saveOrderToDB(record);
    
    if (!emergencyMode) {
        // 非核心逻辑：应急模式下跳过
        sendNotification(record);      // 发送通知
        updateSearchIndex(record);     // 更新搜索索引
        writeAuditLog(record);         // 写审计日志
        callAnalyticsService(record);  // 调用分析服务
    }
}

// 临时消费者组，从 earliest 开始消费所有积压消息
Properties props = new Properties();
props.put("group.id", "emergency-consumer-group-" + System.currentTimeMillis());
props.put("auto.offset.reset", "earliest");  // 从最早开始

// 消费后直接写入临时存储（如 Redis），后续由正常消费者组异步处理
KafkaConsumer<String, String> emergencyConsumer = new KafkaConsumer<>(props);
emergencyConsumer.subscribe(Arrays.asList("order-topic"));

while (lag > 0) {
    ConsumerRecords<String, String> records = emergencyConsumer.poll(Duration.ofMillis(1000));
    for (ConsumerRecord<String, String> record : records) {
        // 快速写入 Redis，不做过多的业务逻辑
        redis.setex("backlog:" + record.key(), 3600, record.value());
    }
}

# 将消费者组的 Offset 重置到最新位置（丢弃所有积压）
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group order-consumer-group \
  --topic order-topic \
  --reset-offsets --to-latest --execute

# 使用 Kafka Connect 将积压消息导出到 HDFS
# 配置 connect-standalone.properties 和 quickstart-hdfs.properties
connect-standalone.sh connect-standalone.properties quickstart-hdfs.properties

分区数 = 峰值生产 TPS / 单分区写入 TPS 上限 × 1.5 (冗余系数)
消费者数 = 分区数
单消费者线程数 = 目标消费 TPS / 单线程处理 TPS

# Kubernetes HPA 示例（基于自定义指标）
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: kafka-consumer-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: order-consumer
  minReplicas: 3
  maxReplicas: 20
  metrics:
  - type: Pods
    pods:
      metric:
        name: kafka_consumer_lag
      target:
        type: AverageValue
        averageValue: "1000"

# 保留 7 天或 100GB
kafka-configs.sh --bootstrap-server localhost:9092 \
  --entity-type topics --entity-name order-topic \
  --alter --add-config retention.ms=604800000,retention.bytes=107374182400

max.poll.records ≤ max.poll.interval.ms / 单条消息处理时间 × 0.8 (安全系数)