本文介绍使用 concurrent.futures 结合 subprocess.run() 并行启动 8 个子进程，为每个进程独立记录微秒级执行时长与退出码（return code），兼顾准确性、可读性与线程安全性。
本文介绍使用 concurrent.futures 结合 subprocess.run() 并行启动 8 个子进程，为每个进程独立记录微秒级执行时长与退出码（return code），兼顾准确性、可读性与线程安全性。
在 Python 中高效管理多个并行子进程，并精确获取其执行时长（建议使用 time.perf_counter() 实现微秒级精度）和返回值（即进程退出码），完全可行且推荐采用现代标准库方案：concurrent.futures + subprocess.run()。相比手动管理 Popen 和 wait()，该组合更简洁、线程安全、异常鲁棒，且天然支持结果聚合。
关键优化点如下：
- ✅ 使用
time.perf_counter()替代datetime.now()：提供更高精度（纳秒级）和单调性，避免系统时钟调整干扰，适合测量短时任务； - ✅ 使用
subprocess.run()而非Popen+wait()：自动阻塞等待完成，并直接返回包含returncode、stdout、stderr的CompletedProcess对象； - ✅ 使用
ThreadPoolExecutor（适用于 I/O 密集型子进程启动/等待）或ProcessPoolExecutor（适用于 CPU 密集型预处理逻辑）；对子进程本身而言，ThreadPoolExecutor已足够——因为subprocess.run()的阻塞本质是 I/O 等待； - ✅ 每个任务函数封装启动、计时、结果提取逻辑，确保每个进程的耗时与返回值严格绑定。
以下为生产就绪示例代码（已修正原答案中 Windows 路径硬编码、shell=True 安全隐患及精度不足问题）：
macOS UI 自动化技能包 —— 用 Python 控制你的 Mac
下载
让 macOS 自动化像说话一样简单 —— 用代码控制你的 Mac，无需学习 AppleScript 系统要求 macOS 14+ (Sonoma) Python 3.9+ 需要开启：辅助功能 + 屏幕录制权限
import concurrent.futures
import subprocess
import time
from typing import Tuple, List, Dict
def run_subprocess(cmd: list, shell: bool = False) -> Tuple[int, float, str]:
"""
执行单个子进程，返回 (returncode, duration_us, cmd_str)
duration_us 为微秒级执行时长（含启动与退出全过程）
"""
start = time.perf_counter()
try:
# 推荐：显式传入 cmd 列表，禁用 shell=True 防注入；如需 shell 功能，请确保输入可信
result = subprocess.run(
cmd,
shell=shell,
capture_output=True,
text=True,
timeout=300 # 防止进程挂起
)
duration_us = int((time.perf_counter() - start) * 1_000_000)
return result.returncode, duration_us, " ".join(cmd) if isinstance(cmd, list) else cmd
except subprocess.TimeoutExpired:
duration_us = int((time.perf_counter() - start) * 1_000_000)
return -1, duration_us, f"TIMEOUT: {' '.join(cmd)}"
except Exception as e:
duration_us = int((time.perf_counter() - start) * 1_000_000)
return -2, duration_us, f"ERROR: {e}"
if __name__ == "__main__":
# 示例命令列表（请替换为你的真实命令，如 ["python", "script.py"]）
commands = [
["sleep", "1"], # Linux/macOS
# ["cmd", "/c", "timeout", "/t", "1"], # Windows
["echo", "hello"],
["date"],
["pwd"],
["ls", "-l"],
["uname", "-a"],
["whoami"],
["hostname"]
]
# 启动最多 8 个并发任务
with concurrent.futures.ThreadPoolExecutor(max_workers=8) as executor:
# 提交所有任务
future_to_cmd = {
executor.submit(run_subprocess, cmd): i
for i, cmd in enumerate(commands)
}
# 按完成顺序收集结果
results: List[Dict] = []
for future in concurrent.futures.as_completed(future_to_cmd):
idx = future_to_cmd[future]
try:
returncode, duration_us, cmd_str = future.result()
results.append({
"index": idx,
"command": cmd_str,
"returncode": returncode,
"duration_us": duration_us
})
print(f"[{idx}] '{cmd_str}' → rc={returncode}, {duration_us:,} μs")
except Exception as exc:
print(f'[{idx}] generated an exception: {exc}')
# 可选：按原始顺序输出汇总（便于调试）
print("\n=== SUMMARY (in submission order) ===")
for res in sorted(results, key=lambda x: x["index"]):
print(f"#{res['index']:2d}: {res['command']:8,} μs")
? 重要注意事项：
-
安全性：避免无条件使用
shell=True，尤其当命令含用户输入时；优先使用cmd列表形式（如["ls", "-l", "/tmp"]）； -
跨平台兼容性：示例中的
sleep、date、uname为 Unix 命令；Windows 用户请改用["cmd", "/c", "ping", "-n", "2", "127.0.0.1"]等等效命令； -
超时控制：务必设置
timeout=参数，防止失控进程永久阻塞线程； -
资源限制：
max_workers=8表示最多 8 个线程并发，但实际子进程数取决于你提交的任务数；若子进程本身开销极大（如启动 JVM），可考虑降低max_workers； -
返回值含义：
returncode=0通常表示成功；非零值需结合具体程序文档解读（如grep找不到匹配时返回 1）。
该方案轻量、可靠、可扩展，完全胜任监控、批处理、CI/CD 工具链等场景下的并行进程时序分析需求。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
