Python多线程执行探测任务并收集结果的异常问题解决方案咨询

该问题由Python闭包延迟绑定的特性导致：你在for循环内定义的lambda表达式引用的probe变量，不会在lambda定义阶段捕获当时的迭代值，只会等到lambda实际运行（线程启动执行）时才查找probe的当前值。此时for循环已经执行完成，probe变量固定指向循环最后一次迭代的probe对象，因此所有线程都会调用同一个probe的run_check方法。

可通过以下两种方案修复：

• 方案1：将当前迭代的probe作为lambda的默认参数传入，默认参数会在函数定义时求值，可捕获当时的变量值

for probe in self.probes.values():
    que = Queue()
    # 新增probe=probe默认参数，绑定当前迭代的probe实例
    t = threading.Thread(target=lambda q, arg1, probe=probe: q.put(probe.run_check(arg1)), args=(que, context))
    jobs.append(t)
    queues.append(que)

• 方案2：使用functools.partial代替lambda，避免闭包绑定问题，代码可读性更高

from functools import partial

def worker(q, run_func):
    q.put(run_func())

for probe in self.probes.values():
    que = Queue()
    # 提前绑定好当前probe的run_check方法和入参context
    run_func = partial(probe.run_check, context)
    t = threading.Thread(target=worker, args=(que, run_func))
    jobs.append(t)
    queues.append(que)

现有原生线程的写法本身支持长耗时任务执行：threading.Thread默认没有执行时长限制，只要run_check方法本身没有异常崩溃，就算执行数分钟也能正常运行并返回结果。但该实现存在两个可优化的风险点：

1. 无超时控制：如果某一个探测任务因为网络故障等原因无限挂死，join()方法会永久等待，导致整个探测流程卡住
2. 手动维护线程、队列的代码冗余，容易出现边界错误

推荐改用concurrent.futures.ThreadPoolExecutor实现，自带任务管理、返回值收集、超时控制能力，代码更简洁易维护：

import time
from functools import partial
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import List

# 省略ProbeResult、ProbeRunnerResult的定义
summary = ProbeRunnerResult()
# 可根据实际情况调整最大并行线程数，避免探针过多时占用过多系统资源
max_workers = min(len(self.probes), 10)

with ThreadPoolExecutor(max_workers=max_workers) as executor:
    # 提交所有探测任务
    future_map = {}
    for probe in self.probes.values():
        run_func = partial(probe.run_check, context)
        future = executor.submit(run_func)
        future_map[future] = probe  # 可保留映射关系，异常时定位具体探针
    
    # 全局超时设置为10分钟，可根据实际需求调整
    try:
        for future in as_completed(future_map.keys(), timeout=600):
            try:
                result = future.result()
                summary.probe_results.append(result)
            except Exception as e:
                # 单个探针执行异常的处理逻辑，可按需扩展
                print(f"探针{future_map[future]}执行失败：{str(e)}")
    except TimeoutError:
        print("部分探针执行超时，已终止等待")

# 补充缺失的结束时间赋值
summary.end_time = time.time()

如果需要保留原有手动管理线程的写法，可以给join()方法增加超时参数，避免永久卡住：

for j in jobs:
    # 单个线程最长等待10分钟
    j.join(timeout=600)
    if j.is_alive():
        # 线程超时后的处理逻辑，比如记录超时结果
        print("某探测任务执行超时")

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