ZH
Tiếng ViệtEnglish한국어Deutsch日本語PortuguêsEspañolFrançais简体中文
首页 / 架构与设计 / Kafka

监控 Kafka：consumer lag 和需要跟踪的关键指标

Cris The Dev · 2026/10/12 · 9 分钟阅读 · 418

1. 概述

一个 Kafka 系统看起来可能“健康”：broker 仍在运行，producer 仍能发送消息。但 consumer 可能在数百万条消息后面，没人知道。反映这种情况的指标是 consumer lag。

在这篇文章中，你将了解 consumer lag 如何计算，以及用三种方法测量它：CLI、Java 中的 AdminClient，以及 Spring Boot 中的 Micrometer。之后，我们会介绍最重要的 broker、producer 和 consumer 指标，以及如何为它们设置告警。

2. Consumer lag 是什么？

Kafka 中的每个 partition 都是一个只追加写入的日志。有两个 offset 你需要关注：

  • Log-end offset (LEO)：下一条要写入 partition 的消息的 offset。
  • Committed offset：consumer group 已提交的 offset，即 group 重启后会继续读取的位置。

一个 partition 的 lag 计算如下：

lag = log-end offset - committed offset

整个 consumer group 的 lag 是该 group 读取的所有 partition 上的 lag 之和。

Lag 等于 0 表示 consumer 已经追上。Lag 轻微波动是正常的，因为消息不断到达。Lag 持续增加 才是令人担忧的信号。这表示 consumer 的处理速度跟不上 producer 的写入速度。

注意： 对于使用事务或压缩的 topic，offset 可能有“间隙”（控制记录、被删除的消息）。因为 lag 按 offset 计算，所以可能会比实际还需要读取的消息数多一点。这不是错误。

3. 通过 CLI 检查 lag

最快的方法是使用 Kafka 附带的脚本：

kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --describe --group order-service

结果如下所示：

GROUP          TOPIC   PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG   CONSUMER-ID       HOST
order-service  orders  0          15230           15232           2     consumer-1-a1b2   /10.0.0.5
order-service  orders  1          14980           18410           3430  consumer-1-c3d4   /10.0.0.6
order-service  orders  2          15102           15102           0     consumer-1-e5f6   /10.0.0.7

Partition 1 落后 3430 条消息，而其他两个 partition 几乎快要追上。Lag 出现这样的差异通常有两个原因。一是某个“热”key 集中进入一个 partition。二是主机 10.0.0.6 上的 consumer 遇到问题。

CLI 在排查问题时非常有用。但你不能每分钟都手动运行这个命令。下一部分将介绍如何自动化 lag 的测量。

4. 使用 AdminClient 计算 lag

添加 kafka-clients 依赖（Spring Boot 已管理版本）：

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
</dependency>

首先，将计算逻辑提取到一个纯函数中以便于测试：

public final class LagCalculator {

    private LagCalculator() {}

    public static Map<TopicPartition, Long> compute(Map<TopicPartition, Long> committed,
                                                    Map<TopicPartition, Long> endOffsets) {
        Map<TopicPartition, Long> lag = new HashMap<>();
        committed.forEach((tp, committedOffset) -> {
            Long end = endOffsets.get(tp);
            if (end != null) {
                lag.put(tp, Math.max(0L, end - committedOffset));
            }
        });
        return lag;
    }
}

这个函数只计算既有 committed offset 又有 end offset 的 partition 的 lag。使用 Math.max(0L, …) 是有原因的。Committed offset 和 end offset 是通过两个不同的请求获取的，所以有时 committed offset 比刚获取的 end offset“更新”。这种情况下的负数无意义，所以我们将其设为 0。

接下来是调用 AdminClient 的类：

public class ConsumerLagChecker {

    private final Admin admin;

    public ConsumerLagChecker(Admin admin) {
        this.admin = admin;
    }

    public Map<TopicPartition, Long> lagOf(String groupId) throws Exception {
        Map<TopicPartition, Long> committed = admin.listConsumerGroupOffsets(groupId)
                .partitionsToOffsetAndMetadata()
                .get(10, TimeUnit.SECONDS)
                .entrySet().stream()
                .filter(e -> e.getValue() != null)
                .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().offset()));

        Map<TopicPartition, OffsetSpec> request = committed.keySet().stream()
                .collect(Collectors.toMap(tp -> tp, tp -> OffsetSpec.latest()));

        Map<TopicPartition, Long> endOffsets = admin.listOffsets(request)
                .all()
                .get(10, TimeUnit.SECONDS)
                .entrySet().stream()
                .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().offset()));

        return LagCalculator.compute(committed, endOffsets);
    }
}

步骤如下：

  • listConsumerGroupOffsets 返回 group 的 committed offset。如果 partition 从未被提交过，值可能为 null，所以我们将其过滤掉。
  • listOffsets 配合 OffsetSpec.latest() 获取每个 partition 的 log-end offset。
  • 每次调用都有 10 秒的超时。这样，当集群出现问题时，线程不会无限期地挂起。

使用方法：

try (Admin admin = Admin.create(Map.of(
        AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"))) {
    new ConsumerLagChecker(admin).lagOf("order-service")
            .forEach((tp, lag) -> System.out.println(tp + " -> " + lag));
}
orders-0 -> 2
orders-1 -> 3430
orders-2 -> 0

4.1. LagCalculator 的单元测试

使用 JUnit 5 和 AssertJ（包含在 spring-boot-starter-test 中）：

class LagCalculatorTest {

    private final TopicPartition p0 = new TopicPartition("orders", 0);
    private final TopicPartition p1 = new TopicPartition("orders", 1);

    @Test
    void shouldComputeLagPerPartition() {
        Map<TopicPartition, Long> lag = LagCalculator.compute(
                Map.of(p0, 100L, p1, 50L),
                Map.of(p0, 120L, p1, 50L));

        assertThat(lag).containsEntry(p0, 20L).containsEntry(p1, 0L);
    }

    @Test
    void shouldClampNegativeLagToZero() {
        Map<TopicPartition, Long> lag = LagCalculator.compute(
                Map.of(p0, 130L),
                Map.of(p0, 120L));

        assertThat(lag).containsEntry(p0, 0L);
    }

    @Test
    void shouldSkipPartitionWithoutEndOffset() {
        Map<TopicPartition, Long> lag = LagCalculator.compute(
                Map.of(p0, 10L), Map.of());

        assertThat(lag).isEmpty();
    }
}

5. 将 lag 导出到 Spring Boot 中的 Micrometer

添加 Actuator 和 Prometheus 注册表：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
<dependency>
    <groupId>io.micrometer</groupId>
    <artifactId>micrometer-registry-prometheus</artifactId>
</dependency>
management.endpoints.web.exposure.include=health,prometheus

5.1. 客户端内置的指标

当有 MeterRegistry 时，Spring Boot 会自动将 MicrometerConsumerListener 附加到 ConsumerFactory。Kafka consumer 的所有内部指标都会出现在 /actuator/prometheus 中，包括：

kafka_consumer_fetch_manager_records_lag_max{client_id="consumer-order-service-1",...} 3430.0
kafka_consumer_fetch_manager_records_lag{topic="orders",partition="1",...} 3430.0

records-lag-max 是这个 consumer 能观察到的最大 lag。Kafka 客户端从 fetch 位置自己计算它，而不是从 committed offset。

5.2. 从外部自测的 Gauge lag

客户端指标有一个很大的弱点。当 consumer 死亡时，指标也会消失。 正好在 lag 增加最快的时候，你却没有了数据。因此，你应该使用 AdminClient 从独立进程中额外测量 lag：

@Component
public class ConsumerLagMetrics {

    private final ConsumerLagChecker checker;
    private final MeterRegistry registry;
    private final Map<TopicPartition, AtomicLong> gauges = new ConcurrentHashMap<>();

    public ConsumerLagMetrics(KafkaAdmin kafkaAdmin, MeterRegistry registry) {
        this.checker = new ConsumerLagChecker(Admin.create(kafkaAdmin.getConfigurationProperties()));
        this.registry = registry;
    }

    @Scheduled(fixedDelay = 30_000)
    public void refresh() throws Exception {
        checker.lagOf("order-service").forEach((tp, lag) ->
                gauges.computeIfAbsent(tp, this::register).set(lag));
    }

    private AtomicLong register(TopicPartition tp) {
        AtomicLong holder = new AtomicLong();
        Gauge.builder("kafka.consumer.group.lag", holder, AtomicLong::get)
                .tag("group", "order-service")
                .tag("topic", tp.topic())
                .tag("partition", String.valueOf(tp.partition()))
                .register(registry);
        return holder;
    }
}

记得在 @Configuration 类上启用 @EnableScheduling。每 30 秒，gauge kafka_consumer_group_lag 就会更新一次。这个值基于 committed offset，所以即使 group 的所有 consumer 都停止了，它仍然是准确的。

在生产环境中，许多团队使用专门的工具（如 kafka-exporter 或 Burrow）来替代上述代码。它们的原理完全相同：比较 committed offset 和 log-end offset。

注意： 如果 consumer service 有多个实例，不要在 consumer service 本身中放置这个 gauge。每个实例都会输出相同的 lag 值，dashboard 聚合后会出错。应该在单独的监控服务中运行它，或仅在一个实例中运行。

6. 需要监控的指标

Lag 告诉你有问题。下面的指标可以帮你找出问题在哪里。

6.1. Broker

指标 (JMX) 含义 期望值
UnderReplicatedPartitions Replica 未同步的 partition 0
OfflinePartitionsCount 没有 leader 的 partition 0
ActiveControllerCount 运行中的 controller 数（整个集群） 正好 1
IsrShrinksPerSec / IsrExpandsPerSec ISR 收缩/扩大 接近 0，不频繁波动
RequestHandlerAvgIdlePercent Request handler 线程的空闲比例 > 0.3
NetworkProcessorAvgIdlePercent Network 线程的空闲比例 > 0.3

OfflinePartitionsCount > 0 是一个严重故障：那个 partition 既不能读也不能写。持续的 UnderReplicatedPartitions 通常表示 broker 过载、磁盘缓慢或网络出现问题。

6.2. Producer

  • record-error-rate：重试耗尽后的发送失败率。任何非零值都值得调查。
  • request-latency-avg：向 broker 的请求延迟。如果增加，你会看到吞吐量下降。
  • record-retry-rate：重试次数多意味着 broker 或网络不稳定。
  • buffer-available-bytes：buffer 耗尽会导致 send() 被阻塞。

6.3. Consumer

  • records-lag-max：已在上面提到。
  • records-consumed-rate：消费速率。与 producer 写入 topic 的速率进行比较。
  • rebalance-rate-per-hour 和 last-rebalance-seconds-ago：持续 rebalance 会导致 consumer“卡住”。这是导致 lag 增加的常见原因。
  • commit-latency-avg：提交缓慢会降低吞吐量，尤其是在同步提交时。
  • last-poll-seconds-ago：如果超过 max.poll.interval.ms，consumer 会被踢出 group。

7. 设置合理的告警

最常见的错误是设置绝对阈值，比如“lag > 1000 就告警”。对于每秒 50,000 条消息的 topic，1000 是无意义的数字。对于每分钟几条消息的 topic，1000 可能已经意味着延迟数小时。

有两种更好的方法。第一种是当 lag 持续增加 时告警：

groups:
  - name: kafka-consumer
    rules:
      - alert: KafkaConsumerLagGrowing
        expr: sum by (group, topic) (deriv(kafka_consumer_group_lag[10m])) > 0
          and sum by (group, topic) (kafka_consumer_group_lag) > 500
        for: 15m
        labels:
          severity: warning
        annotations:
          summary: "{{ $labels.group }} 在 {{ $labels.topic }} 上的 lag 持续增加 15 分钟"

这个规则只在两个条件同时满足时触发：lag 有上升趋势（deriv > 0）且已超过最小阈值。条件必须持续 15 分钟。这样，你可以忽略 deploy 后的短期增长。

第二种方法是将 lag 转换为时间。将 lag 除以消费速率来估算 consumer 落后多少秒。这个数字直接关系到 SLA。例如：“订单必须在 2 分钟内处理”远比“lag 低于 5000”更容易理解。

此外，你应该对指标消失 的情况设置单独的告警，例如使用 absent(kafka_consumer_group_lag)。没有数据并不意味着一切正常。

8. 常见错误

  • 仅依赖 records-lag-max。 当 consumer 死亡时，这个指标会消失。你应该始终有外部的 lag 测量来源。
  • 读取没有活跃 consumer 的 group 的 lag。 CLI 仍会显示 committed offset，但 CONSUMER-ID 列为空。此时 lag 增加意味着没人在处理。
  • 增加 consumer 数超过 partition 数。 多余的 consumer 会闲置，lag 不会减少。要增加并行性，必须增加 partition 数。
  • 忽视 rebalance。 处理单个 batch 用时过长，超过 max.poll.interval.ms。Consumer 被踢出，group rebalance，然后重复。Lag 呈锯齿形增长。应该减少 max.poll.records 或将重处理转移到其他线程。
  • 过于频繁地调用 AdminClient。 每秒查询数百个 group 的 offset 会给 broker 造成不必要的负担。15–60 秒的周期对大多数系统来说就够了。

9. 结论

Consumer lag 是最直接的指标，显示 Kafka 系统是否跟上数据。Lag 通过 log-end offset 减去 committed offset 计算。你可以通过 kafka-consumer-groups.sh 查看 lag，通过 AdminClient 计算，或通过 Micrometer 导出到 Prometheus。

要有效监控，请将客户端指标与独立的测量源相结合。除了 lag，还要监控 broker 指标，如 UnderReplicatedPartitions 和 OfflinePartitionsCount，以及 consumer 端的 rebalance 频率。最后，根据趋势和延迟时间设置告警，而不是根据固定的 lag 数字。这样你既能减少误报，又不会错过真正的问题。

kafkamonitoringspring bootmicrometerprometheus
作者的应用Glowmi · Lantern NightA cozy Mid-Autumn match-3 puzzleiOS · Android

同主题文章