Python实现请求超时后继续执行及计数器方案问询

嘿，你的需求完全可以实现！不过先给你推荐一个更简洁可靠的方案——其实我们常用的requests库（猜你用的是这个）本身就内置了超时控制，比手动写计数器要省心得多。先给你讲这个最直接的方法，再聊聊你提到的计数器思路怎么落地。

方案一：利用requests原生超时参数（推荐） #

requests的请求方法（get/post等）都支持timeout参数，你可以直接设置一个超时时间（单位：秒），如果在这个时间内服务器没响应，就会抛出requests.exceptions.Timeout异常，这样你的代码就不会一直停滞了。

修改后的完整代码 #

import requests
import time

# 假设你的readVoltage()和readCurrent()是已实现的函数
def readVoltage():
    return 0.0
def readCurrent():
    return 0.0

# 主逻辑
for attempt in range(3):
    try:
        print(f'Attempt: {attempt+1}')
        print(f'DUT: {x}')
        time.sleep(1)
        # 这里假设你是POST请求，根据实际情况调整请求方法和参数
        # 设置超时时间为5秒，你可以根据需求修改
        response = requests.post(
            f"http://{ipEntry.get()}{port}",
            json={"action": "temperature_test", "param": x},
            timeout=5
        )
        time.sleep(1)
        print(response.text)
        print(response.json()['result'])  # 假设返回是JSON格式，根据实际结构调整
        answer = response.json()['result']
        with open('log.txt', 'a') as logFile:  # 建议用with语句管理文件，更安全
            logFile.write(f"{time.strftime('%Y_%m_%d-%H_%M_%S')}\t\t{x}\t\t{answer}\t\t{readVoltage()}\t\t{readCurrent()}\n")
    except requests.exceptions.Timeout:
        print(f'Attempt {attempt+1}: 请求超时！')
    except Exception as e:
        print(f'An error occured: {str(e)}')
    else:
        if not answer:
            print(f"Attempt {attempt + 1} failed!")
        else:
            break
else:
    print('All 3 attempts failed')

关键说明： #

• timeout=5表示如果5秒内没收到服务器响应，就触发超时异常
• 我把你原来的response.data.result改成了response.json()['result']，因为requests返回的response默认没有data属性，通常是用json()解析返回的JSON数据，你可以根据实际的响应格式调整
• 用with语句管理日志文件，避免忘记关闭文件导致的问题

方案二：手动计数器+线程实现（针对自定义请求函数） #

如果你用的是自己封装的request函数，不支持原生超时，那可以用线程+计数器的思路：把请求放到一个子线程里执行，主线程同时计时，超过指定时间就终止等待，继续执行后续代码。

实现示例 #

import threading
import time

# 假设你的自定义request函数
def custom_request(url, action, param):
    # 模拟可能阻塞的请求
    time.sleep(10)  # 模拟服务器无响应
    return {"result": True}

# 用于存储请求结果的容器
request_result = None

def request_thread(url, action, param):
    global request_result
    request_result = custom_request(url, action, param)

# 主逻辑
timeout = 5  # 超时时间5秒
for attempt in range(3):
    try:
        print(f'Attempt: {attempt+1}')
        print(f'DUT: {x}')
        time.sleep(1)
        
        # 启动请求线程
        thread = threading.Thread(target=request_thread, args=(f"http://{ipEntry.get()}{port}", "temperature_test", x))
        thread.start()
        
        # 主线程等待，同时检查超时
        start_time = time.time()
        while thread.is_alive():
            if time.time() - start_time > timeout:
                print(f'Attempt {attempt+1}: 请求超时！')
                break
            time.sleep(0.1)  # 避免占用过多CPU
        
        if not request_result:
            raise Exception("请求超时或未返回结果")
        
        time.sleep(1)
        print(request_result)
        answer = request_result['result']
        with open('log.txt', 'a') as logFile:
            logFile.write(f"{time.strftime('%Y_%m_%d-%H_%M_%S')}\t\t{x}\t\t{answer}\t\t{readVoltage()}\t\t{readCurrent()}\n")
    except Exception as e:
        print(f'An error occured: {str(e)}')
    else:
        if not answer:
            print(f"Attempt {attempt + 1} failed!")
        else:
            break
else:
    print('All 3 attempts failed')

注意点： #

• Python的线程无法被强制终止，所以这里的“超时终止”只是主线程不再等待，子线程可能还在后台运行，如果需要严格终止，得在自定义请求函数里加一个停止标志，定期检查
• 这种方式比原生超时复杂，所以优先推荐方案一

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