Python消费Kafka Topic并通过Prometheus HTTP暴露指标遇阻塞问题

你遇到的核心问题是Prometheus的指标收集逻辑和Kafka消费逻辑在同一个执行路径里阻塞了。当Prometheus请求指标端点时，会调用CustomCollector的collect方法，但你的collect方法里写了一个无限循环的Kafka消息消费，这个循环会一直阻塞，根本没法返回指标数据——直到你终止程序，Prometheus才能拿到默认的库指标，这完全符合这个逻辑的表现。

另外你的代码里还有个明显的语法错误：topic = client.topics[b'os.environ['KAFKA_TOPIC'] 这里的字符串拼接和编码逻辑是错的，得先修正。

正确的做法是把Kafka消费放在单独的线程里，让它在后台收集消息统计数据，而collect方法只负责读取已经统计好的数据、生成Prometheus指标返回。这样指标请求不会被消费逻辑阻塞，两者互不干扰。

下面是修正后的完整代码：

import os
import time
import threading
from kafka import KafkaConsumer
from prometheus_client import start_http_server, Metric, REGISTRY

class CustomCollector:
    def __init__(self):
        self.message_count = 0
        # 启动单独的守护线程来消费Kafka消息
        self.consumer_thread = threading.Thread(target=self.consume_kafka, daemon=True)
        self.consumer_thread.start()

    def consume_kafka(self):
        # 初始化Kafka消费者，统一使用kafka库（你之前混用了pykafka和kafka，建议保持一致）
        consumer = KafkaConsumer(
            os.environ['KAFKA_TOPIC'],
            bootstrap_servers=os.environ['KAFKA_ADDRESS'],
            auto_offset_reset='latest'  # 可根据业务需求调整offset策略，比如earliest
        )
        for message in consumer:
            if message is not None:
                print(message.value.decode())  # 解码为字符串方便查看
                self.message_count += 1  # 统计消费的消息总数

    def collect(self):
        # 使用counter类型更符合消息总量的语义（单调递增）
        metric = Metric('kafka_message_consumed_total', 'Total number of Kafka messages consumed', 'counter')
        metric.add_sample('kafka_message_consumed_total', value=self.message_count, labels={})
        yield metric

if __name__ == '__main__':
    start_http_server(9998)
    REGISTRY.register(CustomCollector())
    # 保持主线程持续运行
    while True:
        time.sleep(1)

• 拆分了消费与指标逻辑：把Kafka消费移到独立线程后台执行，避免阻塞Prometheus的指标请求。
• 新增统计变量：用message_count记录消费的消息总数，collect方法仅读取该变量生成指标，全程非阻塞。
• 修正语法与库依赖：修复了topic编码的语法错误，统一使用kafka库的消费者（避免混用不同Kafka客户端带来的兼容性问题）。
• 优化指标类型：将指标改为counter类型，更贴合消息消费总量的统计语义（Prometheus中counter用于单调递增的指标）。

运行修正后的代码后：

1. 控制台依然会正常输出Kafka消息内容；
2. 访问http://localhost:9998/metrics，会立即返回包含kafka_message_consumed_total的指标数据，且数值会随着消息消费自动递增，再也不会出现请求挂起的情况。

内容的提问来源于stack exchange，提问作者korre