Python subprocess管理外部进程的完整实践
Python subprocess管理外部进程的完整实践
subprocess模块替代了os.system、os.popen等旧API。核心是Popen类，run和call是便捷封装。
Popen的基本用法：
import subprocess
proc = subprocess.Popen(['ls', '-la'], stdout=subprocess.PIPE)
output, errors = proc.communicate()
print(output.decode())
Popen启动子进程并立即返回。communicate等待子进程结束并收集输出。
run的便捷封装：
result = subprocess.run(['ls', '-la'], capture_output=True, text=True, check=True)
print(result.stdout)
print(result.returncode)
run等特子进程结束，返回CompletedProcess对象。capture_output=True自动捕获stdout和stderr。text=True以文本模式返回而非字节。
管道连接多个进程：
ls = subprocess.Popen(['ls', '-la'], stdout=subprocess.PIPE)
grep = subprocess.Popen(['grep', 'py'], stdin=ls.stdout, stdout=subprocess.PIPE)
ls.stdout.close()
output = grep.communicate()[0]
print(output.decode())
stdin=ls.stdout将ls的输出连接到grep的输入。ls.stdout.close()在父进程中关闭管道副本，避免死锁。
超时控制：
try:
result = subprocess.run(
['sleep', '10'],
timeout=5,
capture_output=True
)
except subprocess.TimeoutExpired:
print("Process timed out")
timeout=5设置超时。超时后杀死子进程并抛出TimeoutExpired。
环境变量控制：
import os
env = os.environ.copy()
env['MY_VAR'] = 'custom_value'
result = subprocess.run(
['printenv', 'MY_VAR'],
env=env,
capture_output=True,
text=True
)
print(result.stdout) # custom_value
env=dict提供子进程的完整环境变量。省略env继承当前进程的环境。
工作目录：
result = subprocess.run(
['pwd'],
cwd='/tmp',
capture_output=True,
text=True
)
print(result.stdout.strip()) # /tmp
cwd='/tmp'设置子进程的工作目录。
输入重定向：
result = subprocess.run(
['sort'],
input='banana\napple\ncherry\n',
capture_output=True,
text=True
)
print(result.stdout) # apple banana cherry (排序后)
input='string'将字符串作为子进程的stdin。
错误处理：
result = subprocess.run(
['false'],
capture_output=True
)
if result.returncode != 0:
print(f"Process failed with code {result.returncode}")
通过returncode检查进程退出状态。check=True时异常会在非零返回时抛出CalledProcessError。
shell注入防护：
# 正确：参数列表方式（无注入风险）
subprocess.run(['ls', '-la', user_input])
# 错误：字符串方式（有注入风险）
# subprocess.run(f'ls -la {user_input}', shell=True) # 危险
参数列表方式自动转义参数，shell=True时需手动处理user_input。
Popen的其他参数：
proc = subprocess.Popen(
['long_running_process'],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
bufsize=0, # 无缓冲
universal_newlines=True, # 文本模式（Python 3.7+用text=True）
start_new_session=True, # 创建新进程组
creationflags=0 # Windows特定标志
)
start_new_session=True创建新的进程组，子进程及其子孙进程可以一起管理。
进程组管理：
import signal
proc = subprocess.Popen(
['long_running_script.sh'],
start_new_session=True,
)
# 杀死整个进程组
import os
pgid = os.getpgid(proc.pid)
os.killpg(pgid, signal.SIGTERM)
进程组确保子进程创建的孙进程也被终止。
异步Popen：
import asyncio
async def run_command(cmd):
proc = await asyncio.create_subprocess_shell(
cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
stdout, stderr = await proc.communicate()
return stdout, stderr, proc.returncode
asyncio.create_subprocess_exec使用exec方式（推荐），create_subprocess_shell使用shell方式。
资源限制：
import resource
def 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_fn=set_limits # Linux
)
preexec_fn在子进程中fork后exec前执行。可以设置资源限制（RLIMIT_CPU, RLIMIT_AS等）。