如何用Python并行执行多条系统命令并实现失败重试逻辑？

直接上可运行的完整代码，逻辑清晰且满足你的需求：

import subprocess
from concurrent.futures import ThreadPoolExecutor

def run_command(cmd, max_retries=3):
    """执行单个命令，失败时自动重试指定次数"""
    retries = 0
    while retries < max_retries:
        try:
            # 执行命令，捕获标准输出、错误输出和退出码
            result = subprocess.run(
                cmd,
                shell=True,
                check=False,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                text=True
            )
            if result.returncode == 0:
                print(f"✅ 命令执行成功: {cmd}")
                return (cmd, True, result.stdout)
            else:
                retries += 1
                print(f"❌ 命令执行失败（重试 {retries}/{max_retries}）: {cmd}\n错误信息: {result.stderr.strip()}")
        except Exception as e:
            retries += 1
            print(f"⚠️ 命令执行异常（重试 {retries}/{max_retries}）: {cmd}\n异常信息: {str(e)}")
    print(f"❌❌ 命令最终执行失败: {cmd}")
    return (cmd, False, "")

def main():
    # 示例命令列表
    commands = ["ls", "w", "echo 'Hello World'", "some-command-that-may-fail"]
    
    print("开始并行执行命令...")
    # 用线程池实现并行执行
    with ThreadPoolExecutor() as executor:
        results = list(executor.map(run_command, commands))
    
    # 整理执行结果
    successful_cmds = [cmd for cmd, success, _ in results if success]
    failed_cmds = [cmd for cmd, success, _ in results if not success]
    
    print("\n=== 最终执行结果 ===")
    print(f"成功执行: {len(successful_cmds)}个命令")
    for cmd in successful_cmds:
        print(f"- {cmd}")
    print(f"\n执行失败: {len(failed_cmds)}个命令")
    for cmd in failed_cmds:
        print(f"- {cmd}")

if __name__ == "__main__":
    main()

核心逻辑说明 #

• 单命令重试封装：run_command函数单独处理单个命令的执行、错误捕获和重试逻辑，默认最多重试3次，可通过max_retries参数调整。
• 高效并行执行：使用ThreadPoolExecutor实现命令并行，比手动管理Popen进程更简洁，自动处理线程调度，避免重复造轮子。
• 结果清晰区分：执行完成后自动拆分成功/失败命令，方便快速查看执行状态。

对原代码的问题修正 #

1. 原代码用processes.index(process)获取命令索引存在bug：如果命令列表中有重复命令，index会返回第一个匹配项的位置，导致退出码记录错误，现在通过线程池map直接关联命令与结果，彻底避免这个问题。
2. 原代码没有重试逻辑，现在每个失败命令会在函数内部自动重试，无需重新执行所有命令。
3. 添加了标准输出和错误输出的捕获，方便排查命令失败的具体原因。

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