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等）。