ThreadPoolExecutor异常排查：多线程图像处理部分任务未启动

我正在开展一个项目，需处理每日采集的图像及对应Excel数据：对图像执行语义分割、提取目标的高-宽-面积信息并写入对应Excel。单天数据处理耗时约26-32分钟，三天数据总计需1.5-2小时。为提升效率，我使用concurrent.futures.ThreadPoolExecutor编写多线程脚本，期望同时处理三天数据，将总耗时压缩至26-32分钟。

多数情况下脚本运行正常，但偶尔会出现executor.submit()未启动全部3天任务的情况：有时仅启动2天，有时仅启动1天，且无报错信息，该现象随机出现。正常输出会包含三天任务的启动与处理日志，但异常时仅能看到部分天数的处理过程（例如仅day1和day3的任务执行）。

核心代码 #

def processAndsegmentImages(day):
    # 执行图像处理、分割及其他分析
    return '{} images processing and segmentation completed'.format(day)

if __name__ == "__main__":
    start = time.time()
    rf = Roboflow(api_key="my_api_key")
    project = rf.workspace().project("my_project")
    model = project.version(1).model
    print('model loaded\n')
    import concurrent.futures
    days = ['day1', 'day2', 'day3']
    with concurrent.futures.ThreadPoolExecutor(max_workers=min(32, os.cpu_count() + 4)) as executor:
        futures = []
        for day in days:
            print('adding {} to executor'.format(day))
            futures.append(executor.submit(processAndsegmentImages, day=day))
        for future in concurrent.futures.as_completed(futures):
            print(future.result())
    end = time.time()
    tlapsed = end-start
    print('total time taken: {:.2f} minutes'.format(tlapsed/60))

问题原因及解决方案 #

可能原因 #

1. 任务内部静默异常：processAndsegmentImages函数中的图像处理、Excel写入或模型推理操作可能抛出了未被捕获的异常，线程执行时崩溃，但Future对象的异常未被显式处理，导致看起来像是任务未启动。
2. 线程安全问题：共享资源（如Excel文件、Roboflow模型实例）的多线程访问存在竞争，导致某个线程被阻塞或静默失败。例如多线程同时写入同一个Excel文件会引发隐性错误；部分模型实例不支持多线程并发调用，会导致线程执行异常。
3. 资源耗尽阻塞：图像处理、模型推理属于资源密集型操作，若某个线程占用了全部CPU/GPU或内存资源，其他线程无法获取资源执行，表现为任务未启动。

解决方案 #

1. 给任务添加异常捕获：在processAndsegmentImages内部加入异常处理，捕获并输出所有错误，定位问题：

def processAndsegmentImages(day):
    try:
        # 执行图像处理、分割及其他分析
        return '{} images processing and segmentation completed'.format(day)
    except Exception as e:
        error_msg = f"{day} processing failed: {str(e)}"
        print(error_msg)
        return error_msg

同时在遍历Future结果时也添加异常捕获，避免单个任务异常中断整个流程：

for future in concurrent.futures.as_completed(futures):
    try:
        print(future.result())
    except Exception as e:
        print(f"Unexpected error in task: {str(e)}")

2. 确保线程安全：

   • 每个线程处理独立的Excel文件，避免多线程同时写入同一文件；若必须共享文件，使用threading.Lock加锁保护写入操作。
   • 检查Roboflow模型的多线程兼容性：若模型不支持多线程并发，改为每个线程独立加载模型，或改用进程池。

3. 改用ProcessPoolExecutor：Python线程受GIL限制，对CPU/GPU密集型任务的并行效率有限，改用进程池可避免GIL问题，同时每个进程拥有独立的资源空间，减少线程安全风险。注意模型需要在每个进程内初始化：

if __name__ == "__main__":
    start = time.time()
    import concurrent.futures
    days = ['day1', 'day2', 'day3']

    def init_process():
        # 每个进程独立加载模型
        global model
        rf = Roboflow(api_key="my_api_key")
        project = rf.workspace().project("my_project")
        model = project.version(1).model

    with concurrent.futures.ProcessPoolExecutor(max_workers=3, initializer=init_process) as executor:
        futures = [executor.submit(processAndsegmentImages, day=day) for day in days]
        for future in concurrent.futures.as_completed(futures):
            try:
                print(future.result())
            except Exception as e:
                print(f"Task failed: {str(e)}")
    end = time.time()
    print(f'total time taken: {end-start:.2f} minutes')

4. 监控任务状态：在提交任务后，打印每个Future的状态，确认任务是否被调度：

for idx, future in enumerate(futures):
    print(f"Task for {days[idx]} - running: {future.running()}, done: {future.done()}")

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