使用concurrent futures并行执行多文件夹Python脚本无输出问题排查

听起来你遇到了一个典型的并行任务执行陷阱——主程序跑起来没报错，但子文件夹里完全看不到输出，十有八九是子脚本根本没被正确执行，或者执行了但输出到了意料之外的地方。我来帮你拆解几个最可能的原因，以及对应的排查和解决办法：

1. 路径与工作目录问题（最常见） #

很多时候不是子脚本没执行，而是它的输出写去了主脚本的目录，而不是自己所在的子文件夹。这是因为默认情况下，子进程的工作目录和主进程一致，而不是子脚本所在的文件夹。

举个例子，如果你的子脚本里写的是：

with open("output.txt", "w") as f:
    f.write("完成")

那它会把文件写到主脚本运行的目录，而不是子脚本所在的文件夹。

解决办法：
调用子脚本时，指定子进程的工作目录为子脚本所在的文件夹。如果用subprocess配合ProcessPoolExecutor，可以这么写：

import os
import sys
from concurrent.futures import ProcessPoolExecutor
import subprocess

def run_script(script_path):
    script_dir = os.path.dirname(os.path.abspath(script_path))
    try:
        # 用当前主程序的Python解释器执行子脚本，避免环境不一致
        result = subprocess.run(
            [sys.executable, script_path],
            cwd=script_dir,  # 关键：设置工作目录为子脚本所在文件夹
            capture_output=True,
            text=True
        )
        # 打印错误日志，方便排查
        if result.returncode != 0:
            print(f"⚠️ 脚本 {os.path.basename(script_path)} 执行失败: {result.stderr}")
        return result
    except Exception as e:
        print(f"❌ 调用脚本 {script_path} 出错: {str(e)}")
        raise

另外，一定要确保你获取的子脚本路径是绝对路径，不然主程序可能找不到脚本。遍历文件夹时可以用os.path.abspath()来转换：

# 遍历查找目标脚本
target_script_name = "your_script.py"
script_paths = []
for root, _, files in os.walk("."):
    if target_script_name in files:
        full_path = os.path.abspath(os.path.join(root, target_script_name))
        script_paths.append(full_path)
        print(f"找到脚本: {full_path}")  # 打印出来确认路径正确

2. 子脚本缺少执行入口 #

如果你的子脚本里没有if __name__ == "__main__":这个入口，那当你通过某些方式调用它时，核心逻辑可能不会执行。比如如果主程序是用import或者exec()的方式触发子脚本，而不是启动独立进程，那没有入口的话代码就不会跑。

检查方式：手动在子文件夹里执行python your_script.py，看是否能生成输出文件。如果手动执行也没输出，那就是子脚本自身的问题，先把它的入口补全：

# 子脚本 your_script.py
def main():
    # 你的核心逻辑，比如写输出文件
    with open("output.txt", "w") as f:
        f.write("执行成功")

# 必须加这个入口，确保脚本被直接执行时能触发逻辑
if __name__ == "__main__":
    main()

3. 未捕获子任务的异常 #

主程序看起来运行正常，但子任务可能在后台抛出了异常，只是你没看到。比如脚本找不到、依赖缺失、权限不足等问题，都会导致子脚本执行失败，但如果主程序没有捕获这些异常，就会误以为脚本没执行。

解决办法：在主程序中处理Future对象时，一定要调用result()方法，或者用as_completed()来获取结果，这样才能捕获到异常：

def main():
    # ... 前面的脚本查找逻辑 ...
    with ProcessPoolExecutor() as executor:
        futures = [executor.submit(run_script, path) for path in script_paths]
        # 逐个获取任务结果，捕获异常
        for future in futures:
            try:
                future.result()
            except Exception as e:
                print(f"❌ 任务执行失败: {str(e)}")

4. 线程池 vs 进程池的选择问题 #

如果你用的是ThreadPoolExecutor，在Windows下可能会因为GIL（全局解释器锁）的限制，导致子脚本的执行被阻塞，或者某些依赖多进程的逻辑无法正常运行。而且如果子脚本是CPU密集型任务，线程池的效率也极低。

建议：优先用ProcessPoolExecutor来启动独立的Python进程执行子脚本，每个子进程都是独立的环境，避免和主进程的资源冲突。

快速排查步骤 #

1. 手动验证子脚本：去每个子文件夹里手动执行脚本，确认能正常生成输出文件，排除子脚本自身的问题。
2. 打印路径和日志：在主程序中打印找到的所有脚本路径，以及子任务的执行日志，确认脚本被正确找到和调用。
3. 检查工作目录：在子脚本里加一行print(os.getcwd())，看看它的工作目录是不是你预期的子文件夹。
4. 捕获所有异常：确保主程序能捕获到子任务的所有异常，不要让错误“静默”发生。

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