Python实现带超时的上传成功验证（Kafka+Artifactory场景）

我编写了两个函数：

• uploads a gzip file to Artifactory（将gzip文件上传至Artifactory）
• listens to the Kafka topic（监听Kafka主题）

需要在5分钟内通过监听Kafka主题中的关键词push_status: True验证上传是否成功，超时则终止脚本。目前卡在else分支的优雅Python风格实现上，现有代码片段如下：

upload_status = fetch_kafka_topic("topic", "brokers") #从指定主题获取消息
if bool(upload_status) == True: #检查返回的字典是否为空
    if upload_status['file_sha'] == EXPECTED_SHA: #验证SHA值是否与预期一致
        if upload_status['push_status'] == True:
            print("upload is successful ...")
        else:
            # 持续监听Kafka主题5分钟以获取目标关键词，超时则终止构建

优化后的完整实现 #

import time
from kafka import KafkaConsumer  # 假设使用kafka-python库

EXPECTED_SHA = "你的预期SHA值"
KAFKA_TOPIC = "topic"
KAFKA_BROKERS = "brokers"
TIMEOUT = 5 * 60  # 5分钟超时，单位秒

def fetch_kafka_topic(topic, brokers):
    # 原有监听函数，返回单条消息字典（按需调整消息解析逻辑）
    consumer = KafkaConsumer(topic, bootstrap_servers=brokers, auto_offset_reset='latest', consumer_timeout_ms=1000)
    for msg in consumer:
        return eval(msg.value.decode('utf-8'))  # 假设消息为JSON序列化的字典
    return {}

def verify_upload_success():
    # 初始获取消息
    upload_status = fetch_kafka_topic(KAFKA_TOPIC, KAFKA_BROKERS)
    
    # 扁平条件判断，替代多层嵌套if
    if not upload_status:
        print("未获取到Kafka消息，进入超时监听")
    elif upload_status['file_sha'] != EXPECTED_SHA:
        print(f"SHA校验失败：预期{EXPECTED_SHA}，实际{upload_status['file_sha']}")
    elif upload_status['push_status']:
        print("upload is successful ...")
        return True
    
    # 超时监听逻辑
    start_time = time.time()
    consumer = KafkaConsumer(KAFKA_TOPIC, bootstrap_servers=KAFKA_BROKERS, auto_offset_reset='latest')
    
    while time.time() - start_time < TIMEOUT:
        for msg in consumer:
            try:
                upload_status = eval(msg.value.decode('utf-8'))
                # 同时校验SHA和push_status
                if upload_status.get('file_sha') == EXPECTED_SHA and upload_status.get('push_status'):
                    print("upload is successful ...")
                    consumer.close()
                    return True
            except Exception as e:
                print(f"解析Kafka消息失败：{e}")
                continue
        time.sleep(1)  # 避免空循环占用过高CPU
    
    # 超时终止处理
    print("监听超时，上传验证失败，终止构建")
    consumer.close()
    return False

# 执行验证
verify_upload_success()

关键优化点 #

• 扁平条件结构：用elif替代多层嵌套if，代码可读性大幅提升
• 精确超时控制：通过time.time()计算流逝时间，实现严格的5分钟超时逻辑
• 资源复用：监听阶段复用Kafka消费者实例，避免重复创建连接
• 健壮性增强：增加消息解析的异常捕获，避免单条错误消息导致脚本崩溃
• CPU友好：循环中加入1秒休眠，减少空循环的资源占用

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