4、进程管理与并发执行：使用subprocess模块运行外部命令、捕获输出、处理错误

说实话，很多人在写脚本时都踩过同一个坑——用 os.system() 跑个命令，然后眼巴巴等着结果。但你要真想做好批量脚本的自动化，subprocess 才是你的正牌搭档。我最早接触它是在一次线上部署任务中，几十台机器要同时执行升级脚本，用 os.system 搞得我焦头烂额。后来换了 subprocess，整个世界都清净了。

4.1 为什么是 subprocess？

说白了，subprocess 就是 Python 官方推荐的「跑外部命令」的标准方式。它比 os.system 强在哪？我给你列几个关键点：

  • 捕获输出：你能拿到 stdout 和 stderr，而不是只能看打印
  • 控制输入：可以通过 stdin 向子进程喂数据
  • 超时控制：跑死了能掐掉，不会一直挂在那
  • 并发执行：配合多进程，批量起飞

嗯，这里要注意：subprocess 的核心函数就两个——run() 和 Popen()。前者是高级封装，适合大多数场景；后者是底层接口，灵活但需要你多操点心。

4.2 基础用法：run() 函数

我个人习惯，90% 的场景用 run() 就够了。看个例子：

import subprocess

result = subprocess.run(
    ['ls', '-l', '/tmp'],
    capture_output=True,
    text=True,
    timeout=10
)

print(result.stdout)   # 标准输出
print(result.stderr)   # 错误输出
print(result.returncode)  # 返回码，0 表示成功

我在项目中遇到过一个问题：有人用 shell=True 传字符串命令，结果参数里带空格就炸了。我的建议是——尽量用列表传参，安全又可控。

小技巧：设置 text=True 后，输出直接是字符串，不用自己 decode。Python 3.7+ 都支持，省心。

4.3 捕获输出与错误处理

你想想看，批量跑脚本时最怕什么？怕某个命令悄悄失败了，你还不知道。所以捕获输出和错误处理是必修课。

我一般这样处理：

def run_command(cmd, timeout=30):
    try:
        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=timeout,
            check=False  # 不自动抛异常，自己控制
        )
        if result.returncode != 0:
            print(f"[WARN] 命令失败: {cmd}")
            print(f"错误信息: {result.stderr}")
            return None
        return result.stdout.strip()
    except subprocess.TimeoutExpired:
        print(f"[ERROR] 命令超时: {cmd}")
        return None
    except Exception as e:
        print(f"[ERROR] 未知错误: {e}")
        return None
注意：check=True 虽然方便，但一旦返回码非零就会抛 CalledProcessError。如果你要自己处理错误，建议用 check=False 手动判断。

我曾经因为没捕获 stderr，排查了整整一个下午。后来发现是某个脚本偷偷往 stderr 里写警告信息，而我只看了 stdout。所以现在我的原则是——两个流都抓，一个别漏。

4.4 并发执行：让脚本飞起来

单线程跑脚本，遇到 I/O 密集任务时效率极低。比如你要在 100 台机器上执行 uptime，串行跑要等 100 次网络往返。这时候并发就派上用场了。

我常用的模式是 subprocess.Popen + concurrent.futures：

import subprocess
from concurrent.futures import ThreadPoolExecutor, as_completed

def run_remote(host, command):
    ssh_cmd = ['ssh', host, command]
    proc = subprocess.Popen(
        ssh_cmd,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True
    )
    stdout, stderr = proc.communicate(timeout=30)
    return host, stdout.strip(), stderr.strip(), proc.returncode

hosts = ['server1', 'server2', 'server3']
commands = ['uptime', 'df -h', 'free -m']

with ThreadPoolExecutor(max_workers=5) as executor:
    futures = {executor.submit(run_remote, h, c): h for h, c in zip(hosts, commands)}
    for future in as_completed(futures):
        host, out, err, code = future.result()
        print(f"{host}: {out}")
核心思路：用 Popen 启动进程，用 communicate() 等待结束并获取输出，用线程池控制并发数。这样既不会把机器跑死，又能大幅提升效率。

这里有个坑——Popen 如果不调 communicate() 或 wait()，子进程会变成僵尸进程。我刚开始用的时候没注意，结果服务器上挂了一堆 defunct 进程，被运维大哥骂了一顿。嗯，从那以后我每次都用 communicate() 收尾。

4.5 知识体系：一张图说清楚

下面这张 SVG 图，帮你理清 subprocess 的核心脉络：

subprocess 模块核心知识体系 subprocess run() 高级接口 capture_output timeout 超时控制 check 返回码检查 Popen() 底层接口 communicate() 交互 并发 + ThreadPool run() 适合简单场景，Popen() 适合复杂并发控制

4.6 实战避坑指南

最后，分享几个我踩过的坑，希望能帮你省点时间：

  • 管道死锁：如果你用 Popen 且 stdout/stderr 都是 PIPE，但没及时读取，缓冲区满了就会死锁。解决方案——用 communicate() 一次性读完，或者用 subprocess.PIPE 配合线程读。
  • Shell 注入：别把用户输入直接拼到命令里。用列表传参，Python 会自动转义。
  • 环境变量：子进程默认继承父进程的环境。如果你要隔离，可以传 env={} 或自定义字典。
  • 信号处理：批量杀子进程时，记得用 proc.terminate() 或 proc.kill()，别用 os.kill() 乱搞。
我的习惯：写批量脚本时，我会先写一个 run_cmd() 包装函数，把超时、重试、日志都封装进去。这样后面每个任务都调同一个函数，出问题也好排查。

好了，关于 subprocess 的进程管理和并发执行，就聊到这儿。记住一句话：能用 run() 别用 Popen()，能用列表别用字符串，能捕获输出就别偷懒。这样写出来的脚本，既稳当又高效。

可可精品资料库

← 上一章 📖 返回目录 下一章 →