本文介绍使用 concurrent.futures 配合 subprocess.run() 并行启动 8 个子进程，为每个进程独立记录微秒级执行时长与退出码（return code），兼顾准确性、可读性与线程安全性。
本文介绍使用 concurrent.futures 配合 subprocess.run() 并行启动 8 个子进程，为每个进程独立记录微秒级执行时长与退出码（return code），兼顾准确性、可读性与线程安全性。
在 Python 中高效管理并行子进程，并精确获取每个进程的执行耗时（建议用 time.perf_counter_ns() 达到纳秒级精度，可换算为微秒）和返回值（即退出码），关键在于避免阻塞式等待（如 Popen.wait()），转而采用异步提交 + 结果聚合的模式。推荐使用 concurrent.futures.ThreadPoolExecutor（适用于 I/O 密集型子进程启动场景）或 ProcessPoolExecutor（适用于 CPU 密集型任务预处理），配合 subprocess.run() —— 它比 Popen 更简洁安全，能自动捕获完成状态、标准输出/错误及返回码。
以下是一个生产就绪的示例，支持：
- 并行启动最多 8 个子进程；
- 对每个进程单独计时（起始至完全退出，单位：微秒）；
- 获取真实返回值（
CompletedProcess.returncode）； - 容错处理（如超时、异常）；
- 清晰的结果结构化输出。
import concurrent.futures
import subprocess
import time
from typing import Tuple, Dict, Any
def run_subprocess_with_timing(cmd: str, timeout: float = 30.0) -> Dict[str, Any]:
"""
执行单个子进程，返回带时间戳和返回值的结构化结果
:param cmd: 要执行的命令字符串（支持 shell=True）
:param timeout: 最大允许运行时间（秒），超时将终止进程
:return: 包含命令、返回码、耗时（微秒）、是否超时的字典
"""
start_ns = time.perf_counter_ns() # 纳秒级高精度起点
try:
result = subprocess.run(
cmd,
shell=True,
universal_newlines=True,
stderr=subprocess.STDOUT,
stdout=subprocess.PIPE,
timeout=timeout
)
end_ns = time.perf_counter_ns()
elapsed_us = (end_ns - start_ns) // 1000 # 转为微秒（整数）
return {
"command": cmd,
"returncode": result.returncode,
"duration_us": elapsed_us,
"stdout": result.stdout.strip() if result.stdout else "",
"timed_out": False
}
except subprocess.TimeoutExpired as e:
end_ns = time.perf_counter_ns()
elapsed_us = (end_ns - start_ns) // 1000
return {
"command": cmd,
"returncode": -1,
"duration_us": elapsed_us,
"stdout": f"TIMEOUT after {timeout}s",
"timed_out": True
}
except Exception as e:
end_ns = time.perf_counter_ns()
elapsed_us = (end_ns - start_ns) // 1000
return {
"command": cmd,
"returncode": -2,
"duration_us": elapsed_us,
"stdout": f"ERROR: {str(e)}",
"timed_out": False
}
# 主执行逻辑
if __name__ == "__main__":
# 示例命令列表（请替换为你的真实命令，如 ["python script1.py", "ls -l", ...]）
commands = [f"sleep {i % 3 + 1}" for i in range(8)] # 模拟不同耗时的命令
results = []
with concurrent.futures.ThreadPoolExecutor(max_workers=8) as executor:
# 提交所有任务
future_to_cmd = {
executor.submit(run_subprocess_with_timing, cmd): cmd
for cmd in commands
}
# 按完成顺序收集结果
for future in concurrent.futures.as_completed(future_to_cmd):
try:
result = future.result()
results.append(result)
print(f"[✓] '{result['command']}' → rc={result['returncode']}, "
f"duration={result['duration_us']} μs")
except Exception as exc:
print(f'[✗] Task generated exception: {exc}')
# 可选：汇总统计
print("\n--- SUMMARY ---")
print(f"Total processes: {len(results)}")
successful = sum(1 for r in results if r["returncode"] == 0)
print(f"Successful: {successful}/{len(results)}")
if results:
avg_us = sum(r["duration_us"] for r in results) // len(results)
max_us = max(r["duration_us"] for r in results)
print(f"Average duration: {avg_us} μs | Max: {max_us} μs")✅ 关键优势说明：
macOS UI 自动化技能包 —— 用 Python 控制你的 Mac
下载
让 macOS 自动化像说话一样简单 —— 用代码控制你的 Mac，无需学习 AppleScript 系统要求 macOS 14+ (Sonoma) Python 3.9+ 需要开启：辅助功能 + 屏幕录制权限
- 使用
time.perf_counter_ns()替代datetime.now()或time.time()，避免系统时钟调整干扰，提供单调、高精度计时； -
subprocess.run()自动阻塞直至进程结束，并原生返回returncode，无需手动p.wait()+p.returncode； -
ThreadPoolExecutor天然支持并发控制（max_workers=8）与结果顺序无关的异步收集（as_completed）； - 内置超时与异常处理，防止某个卡死进程拖垮整体流程；
- 返回结构化字典，便于后续日志分析、JSON 序列化或监控集成。
⚠️ 注意事项：
立即学习“Python免费学习笔记（深入）”；
- 若子进程本身是 CPU 密集型（如科学计算），建议改用
concurrent.futures.ProcessPoolExecutor避免 GIL 争用； -
shell=True存在安全风险，请确保cmd来源可信；生产环境推荐shell=False+ 参数列表形式（如["ls", "-l", "/tmp"]）； - Windows 下
sleep命令需替换为timeout /t N >nul，或统一使用跨平台工具（如 Python 的time.sleep()封装脚本）； -
DEVNULL在示例中未显式使用，但你可在subprocess.run(..., stdout=subprocess.DEVNULL)中关闭输出以提升性能。
掌握这一模式后，你不仅能精准度量多进程性能，还可轻松扩展为健康检查、批量测试、CI/CD 任务编排等实际工程场景。
