在日常运维中，Kafka 集群的巡检是保障消息队列稳定运行的重要环节。但当集群规模扩大到 1600 + 个 topic 时，单线程巡检耗时长达 2 小时，严重影响效率。本文分享如何通过 Python 多线程并发改造，将耗时压缩至 9 分钟，同时保证 Prometheus 服务稳定。

一、背景：单线程巡检的痛点

我们的 Kafka 巡检任务需要对每个 topic 查询 3 个关键指标（消息写入速率的最大值、最小值、平均值），涉及 1622 个 topic，累计查询次数超 5000 次。原单线程代码流程如下：

1. 遍历所有 Kafka 集群，获取每个集群的 topic 列表
2. 对每个 topic 串行执行 3 次 Prometheus 查询（max/min/avg）
3. 汇总结果并上报 CMDB

问题：单线程串行执行导致总耗时长达 2 小时，且中间任何一次查询阻塞都会拖慢整体进度。

二、改造思路：多线程并发优化

针对 IO 密集型的网络请求（Prometheus 查询），多线程是最佳选择。核心改造思路如下：

1. 任务拆分：将单个 topic 的 3 次查询封装为独立任务
2. 并发控制：使用ThreadPoolExecutor限制最大并发数（避免压垮 Prometheus）
3. 异常处理：为每个请求添加超时控制，捕获失败任务不影响整体
4. 结果聚合：异步收集所有任务结果，统一上报

三、关键代码改造

3.1 单 topic 处理函数（核心任务单元）

将单个 topic 的 3 次 Prometheus 查询封装为函数，返回结构化结果（失败返回 None）：

python

运行

import requests
import datetime
from typing import Dict, Optional

def process_single_topic(job: str, server: str, topic: str, partition: int) -> Optional[Dict]:
    """处理单个topic的三次Prometheus查询，返回组装后的结果"""
    try:
        # 1. 查询写入速率最大值（max）
        max_query = (
            f'max_over_time(sum(delta(kafka_topic_partition_current_offset{{job="{job}",server="{server}",topic="{topic}"}}[1m])/60) '
            f'by (job,server,topic)[1d:2m])'
        )
        max_response = requests.get(
            PROMETHEUS_URL.replace('query_range', 'query'),
            params={'query': max_query},
            timeout=15  # 超时控制：15秒
        )
        max_data = max_response.json()
        if not max_data.get('data') or not max_data['data'].get('result'):
            print(f"[Warning] topic {topic} 无max数据，跳过")
            return None
        message_write_rate_max = round(float(max_data['data']['result'][0]['value'][1]), 2)

        # 2. 查询写入速率最小值（min）
        # 省略min数据解析（类似max逻辑）

        # 3. 查询写入速率平均值（avg）

        # 省略avg数据解析（类似max逻辑）

        # 组装结果
        return {
            'topic_day': f'{topic}_{inspect_day}',
            'topic': topic,
            'partition': partition,
            'cluster_name': job,
            'message_write_rate_max': message_write_rate_max,
            'message_write_rate_min': message_write_rate_min,
            'message_write_rate_avg': message_write_rate_avg,
            'time': datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        }

    except Exception as e:
        print(f"处理topic {topic} 失败: {str(e)}")
        return None

3.2 多线程并发调度逻辑

使用线程池并发执行所有 topic 任务，控制最大并发数为 10：

python

运行

from concurrent.futures import ThreadPoolExecutor, as_completed

def get_topic_inspect():
    """多线程并发处理所有topic巡检"""
    job_server_dict = get_kafka_clusters()  # 获取Kafka集群列表
    if not job_server_dict:
        print("无Kafka集群信息，退出巡检")
        return

    cmdb_request = CmdbRequest()  # 初始化CMDB请求工具

    for job, server in job_server_dict.items():
        # 1. 获取当前集群的所有topic及分区数
        ...
        # 2. 构建topic→partition映射
        topic_dict =...
        print(f"集群 {job} 共{len(topic_dict)}个topic，开始并发处理...")

        # 3. 10个并发处理所有topic
        topic_inspect_list = []
        with ThreadPoolExecutor(max_workers=10) as executor:  # 控制并发数
            # 提交所有任务到线程池
            futures = [
                executor.submit(process_single_topic, job, server, topic, partition)
                for topic, partition in topic_dict.items()
            ]
            # 异步收集结果
            for future in as_completed(futures):
                result = future.result()
                if result:
                    topic_inspect_list.append(result)

        # 4. 汇总结果上报CMDB
        print(f"集群 {job} 处理完成，共{len(topic_inspect_list)}条有效数据")
        if topic_inspect_list:
            topic_list_post_data = cmdb_request.post(
                objectId='KAFKA_TOPIC_INSPECT',
                keys=['topic_day'],
                datas=topic_inspect_list
            )
            print(f"集群 {job} 上报结果: {topic_list_post_data}")

    print("所有Kafka集群的topic巡检完成")

四、改造效果：从 2 小时到 9 分钟

优化前后对比：

• 单线程：2 小时 +（5000 + 次查询串行执行）
• 10 线程并发：9 分钟（平均每次查询耗时～0.1 秒，并发高效利用网络等待时间）

执行日志片段：

plaintext

[Debug] 集群列表: ['Kafka-elk_test1', 'flume-es-3', 'jrez-3'...]
集群 Kafka-elk_test1 共361个topic，开始并发处理...
集群 Kafka-elk_test1 处理完成，共361条有效数据
----CMDB上报结果: 
{'code': 0, 'error': '', 'message': '', 'data': {'insert_count': 0, 'update_count': 361, 'failed_count': 0, 'data': []}}
...
所有Kafka集群的topic巡检完成

五、注意事项

1. 并发数控制：根据 Prometheus 性能调整max_workers（建议从 10 开始，逐步测试最大承受能力），避免因请求过多导致 Prometheus 超时。
2. 超时与重试：为每个请求添加timeout（10-15 秒），关键场景可增加重试逻辑（如tenacity库）。
3. 资源监控：改造后需观察 Prometheus 的query_duration_seconds（查询延迟）和http_requests_total（错误率），确保服务稳定。
4. 异常隔离：单个 topic 处理失败不影响其他任务，通过日志记录失败详情，便于后续排查。

六、总结

对于 IO 密集型的批量查询任务，Python 多线程是简单高效的优化手段。本文通过ThreadPoolExecutor将 Kafka 巡检耗时从 2 小时压缩至 9 分钟，既保证了效率提升，又通过并发控制和异常处理保障了稳定性。

如果你也面临类似的批量查询效率问题，不妨尝试这种改造思路，欢迎在评论区交流你的实践经验！