如何在Python中限制调用外部命令的子进程并发数量？

针对你用subprocess.Popen批量启动mayabatch.exe处理ma文件的场景，有两种可靠的方式控制并发进程数量：

1. 手动维护活跃进程列表（适合理解底层逻辑） #

核心思路是维护一个正在运行的进程列表，每次启动新进程前检查列表长度，若已达并发上限，就等待列表中某个进程结束后再启动新进程。

import os
import subprocess

def process_ma_file(file_path):
    # 启动mayabatch.exe进程，根据实际需求调整参数
    return subprocess.Popen(
        ["mayabatch.exe", "-file", file_path],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True
    )

def main():
    target_dir = "你的ma文件文件夹路径"
    max_concurrent = 2  # 限制同时运行2个进程
    active_processes = []

    # 遍历文件夹中的ma文件
    for filename in os.listdir(target_dir):
        if filename.endswith(".ma"):
            file_path = os.path.join(target_dir, filename)
            
            # 若活跃进程数已达上限，等待任意一个进程结束
            while len(active_processes) >= max_concurrent:
                # 移除已结束的进程
                active_processes = [p for p in active_processes if p.poll() is None]
                # 若仍满额，短暂等待后再检查
                if len(active_processes) >= max_concurrent:
                    subprocess.wait(active_processes, timeout=1)
            
            # 启动新进程并加入活跃列表
            proc = process_ma_file(file_path)
            active_processes.append(proc)
            print(f"启动进程处理文件: {filename}")
    
    # 等待最后一批进程全部结束
    for proc in active_processes:
        proc.wait()
        stdout, stderr = proc.communicate()
        if stdout:
            print(f"文件{proc.args[2]}处理输出:\n{stdout}")
        if stderr:
            print(f"文件{proc.args[2]}处理错误:\n{stderr}")

if __name__ == "__main__":
    main()

2. 使用concurrent.futures.ThreadPoolExecutor（推荐，代码更简洁） #

线程池会自动帮你管理并发数，无需手动维护进程列表，每个线程负责启动一个子进程并等待其完成。

import os
import subprocess
from concurrent.futures import ThreadPoolExecutor

def process_ma_file(file_path):
    print(f"开始处理文件: {os.path.basename(file_path)}")
    # 启动进程并等待完成，获取输出结果
    result = subprocess.run(
        ["mayabatch.exe", "-file", file_path],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
        check=False
    )
    if result.returncode == 0:
        print(f"文件{os.path.basename(file_path)}处理成功")
    else:
        print(f"文件{os.path.basename(file_path)}处理失败，错误信息:\n{result.stderr}")
    return result

def main():
    target_dir = "你的ma文件文件夹路径"
    max_concurrent = 2  # 限制并发数
    ma_files = [os.path.join(target_dir, f) for f in os.listdir(target_dir) if f.endswith(".ma")]

    # 用线程池批量执行任务
    with ThreadPoolExecutor(max_workers=max_concurrent) as executor:
        executor.map(process_ma_file, ma_files)

if __name__ == "__main__":
    main()

注意事项 #

• 两种方式都可根据实际需求调整max_concurrent的值；
• 若mayabatch.exe是CPU密集型任务，线程池依然适用——因为子进程是独立运行的，线程池仅负责调度启动逻辑；
• 若处理的ma文件输出内容较多，建议将stdout/stderr重定向到本地文件，避免内存占用过高。

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