Python subprocess管理外部进程的完整实践subprocess模块替代了os.system、os.popen等旧API。核心是Popen类run和call是便捷封装。Popen的基本用法import subprocessproc subprocess.Popen([ls, -la], stdoutsubprocess.PIPE)output, errors proc.communicate()print(output.decode())Popen启动子进程并立即返回。communicate等待子进程结束并收集输出。run的便捷封装result subprocess.run([ls, -la], capture_outputTrue, textTrue, checkTrue)print(result.stdout)print(result.returncode)run等特子进程结束返回CompletedProcess对象。capture_outputTrue自动捕获stdout和stderr。textTrue以文本模式返回而非字节。管道连接多个进程ls subprocess.Popen([ls, -la], stdoutsubprocess.PIPE)grep subprocess.Popen([grep, py], stdinls.stdout, stdoutsubprocess.PIPE)ls.stdout.close()output grep.communicate()[0]print(output.decode())stdinls.stdout将ls的输出连接到grep的输入。ls.stdout.close()在父进程中关闭管道副本避免死锁。超时控制try:result subprocess.run([sleep, 10],timeout5,capture_outputTrue)except subprocess.TimeoutExpired:print(Process timed out)timeout5设置超时。超时后杀死子进程并抛出TimeoutExpired。环境变量控制import osenv os.environ.copy()env[MY_VAR] custom_valueresult subprocess.run([printenv, MY_VAR],envenv,capture_outputTrue,textTrue)print(result.stdout) # custom_valueenvdict提供子进程的完整环境变量。省略env继承当前进程的环境。工作目录result subprocess.run([pwd],cwd/tmp,capture_outputTrue,textTrue)print(result.stdout.strip()) # /tmpcwd/tmp设置子进程的工作目录。输入重定向result subprocess.run([sort],inputbanana\napple\ncherry\n,capture_outputTrue,textTrue)print(result.stdout) # apple banana cherry (排序后)inputstring将字符串作为子进程的stdin。错误处理result subprocess.run([false],capture_outputTrue)if result.returncode ! 0:print(fProcess failed with code {result.returncode})通过returncode检查进程退出状态。checkTrue时异常会在非零返回时抛出CalledProcessError。shell注入防护# 正确参数列表方式无注入风险subprocess.run([ls, -la, user_input])# 错误字符串方式有注入风险# subprocess.run(fls -la {user_input}, shellTrue) # 危险参数列表方式自动转义参数shellTrue时需手动处理user_input。Popen的其他参数proc subprocess.Popen([long_running_process],stdinsubprocess.PIPE,stdoutsubprocess.PIPE,stderrsubprocess.PIPE,bufsize0, # 无缓冲universal_newlinesTrue, # 文本模式Python 3.7用textTruestart_new_sessionTrue, # 创建新进程组creationflags0 # Windows特定标志)start_new_sessionTrue创建新的进程组子进程及其子孙进程可以一起管理。进程组管理import signalproc subprocess.Popen([long_running_script.sh],start_new_sessionTrue,)# 杀死整个进程组import ospgid os.getpgid(proc.pid)os.killpg(pgid, signal.SIGTERM)进程组确保子进程创建的孙进程也被终止。异步Popenimport asyncioasync def run_command(cmd):proc await asyncio.create_subprocess_shell(cmd,stdoutasyncio.subprocess.PIPE,stderrasyncio.subprocess.PIPE)stdout, stderr await proc.communicate()return stdout, stderr, proc.returncodeasyncio.create_subprocess_exec使用exec方式推荐create_subprocess_shell使用shell方式。资源限制import resourcedef set_limits():resource.setrlimit(resource.RLIMIT_CPU, (1, 1))resource.setrlimit(resource.RLIMIT_AS, (100 * 1024 * 1024, 100 * 1024 * 1024))proc subprocess.Popen([some_command],preexec_fnset_limits # Linux)preexec_fn在子进程中fork后exec前执行。可以设置资源限制RLIMIT_CPU, RLIMIT_AS等。