Kafka Consumer 健康检查：死或活

Source: Dev.to

许多人都有过这样的经历——凌晨 3 点，手机震动，你看到一条警报：Kafka 消费者延迟超过阈值。
你慌忙打开笔记本，查看指标，结果……一切看起来都正常。消息正在被处理。延迟只是下游服务慢导致的短暂峰值。于是你关闭警报，回床继续睡觉，心里又多了一个“以后再修复”的待办项。

听起来很熟悉吧？

实际上发生的情况是：延迟只能告诉你落后了多少，但不能说明你是否在前进。一个消费者可能因为卡住而在 1 000 条消息的延迟上停留 10 分钟，也可能正好以消息到达的速度处理，保持同样的延迟。仅凭延迟，你无法区分这两种情况。

真正的问题不是“延迟多少？”——而是“我们在前进吗？”

---

为什么传统健康检查不足

“永远健康”方式

func healthHandler(w http.ResponseWriter, r *http.Request) {
    w.WriteHeader(http.StatusOK)
    w.Write([]byte("OK")) // "I'm alive!" (Are you though?)
}

这根本没有提供任何信息。你的消费者可能已经完全卡死，而 Kubernetes（或任何编排系统）仍会把它当作活着的实例继续运行。

“Ping Broker”方式

检查连通性总比什么都不做好，但连通性并不等同于消费者在处理消息。网络可能正常，而消费者组却陷入无限的再平衡循环。

“延迟阈值”陷阱

大多数团队通过 Prometheus 或类似工具监控消费者延迟，并在延迟超过阈值时触发警报——阈值可能是 100 条、1 000 条，甚至是百万条。

这个阈值应该设多少合适？

• 设得太低 → 你会在凌晨 3 点被叫醒，因为下游 API 响应慢了 30 秒。消费者本身没问题，只是短暂超负荷。误报。
• 设得太高 → 真正的问题会被忽视，直到它们已经演变成面向客户的故障。等警报触发时，你已经进入抢修模式。

粒度问题：在 50 个副本的部署中，单个卡住的 Pod 可能被平均延迟掩盖而不被发现。该 Pod 静默失败，而其他 49 个正常工作，导致聚合指标看不出异常。

根本的矛盾在于：快速检测会产生误报，可靠的警报则会导致检测延迟。使用复杂的启发式或机器学习模型只会增加复杂度，却仍然无法实现即时发现。

核心洞见：进度 vs. 位置

我们真正需要的是一种方式来回答一个简单的问题：这个特定的消费者实例在前进吗？

与其测量延迟，不如测量进度。

1. 心跳 – 记录每个分区最近一次处理的消息时间戳。如果我们在处理消息，就说明健康。
2. 验证 – 如果在一定时间内（比如 X 秒）没有新消息，我们查询 Kafka Broker 的最新偏移量。

这样可以清晰地区分卡住的消费者和空闲的消费者。

工作原理：三种场景

场景 1：活跃处理（健康）

当消息不断流入且消费者正在处理时，健康检查是即时完成的——不需要查询 Broker，因为已经观察到最近的活动。

场景 2：卡住的消费者（不健康）

消费者冻结，而消息仍在不断到达。查询 Broker 后发现还有未处理的工作，但消费者没有在消费。

场景 3：空闲的消费者（健康）

消费者已经追上所有消息，正等待新消息的到来。

相同的超时时间会根据 Broker 的状态产生不同的结果。

实现方式

我已经把这套逻辑封装进 kafka-pulse-go，这是一个轻量级库，兼容大多数流行的 Kafka 客户端。核心逻辑与具体客户端实现解耦，并提供以下适配器：

• Sarama
• segmentio/kafka-go
• Confluent 的客户端

配置监控

下面是使用 Sarama 的最小示例：

import (
    "log"
    "time"

    "github.com/IBM/sarama"
    adapter "github.com/vmyroslav/kafka-pulse-go/adapter/sarama"
    "github.com/vmyroslav/kafka-pulse-go/pulse"
)

func main() {
    // 你的现有 Sarama 配置
    config := sarama.NewConfig()
    client, err := sarama.NewClient([]string{"broker1:9092", "broker2:9092"}, config)
    if err != nil {
        log.Fatal(err)
    }

    // 创建健康检查适配器
    brokerClient := adapter.NewClientAdapter(client)

    // 配置监控器
    monitorConfig := pulse.Config{
        Logger:       log.Default(),
        StuckTimeout: 30 * time.Second,
    }

    monitor, err := pulse.NewHealthChecker(monitorConfig, brokerClient)
    if err != nil {
        log.Fatal(err)
    }

    // 将 `monitor` 注入你的消费者，并暴露其健康检查端点
}

StuckTimeout 是什么？
它定义了监控器在未看到新消息时等待多长时间才去查询 Broker。该值应依据你的主题消息频率来设定：

• 高吞吐量主题：10–30 秒
• 中等流量主题：1–2 分钟
• 低流量主题：5–10 分钟

调整此超时时间可以在灵敏度（快速检测）和误报（短暂停顿）之间取得平衡。

💡 想直接查看代码实现？完整源码请访问 这里。