企业官网建设流程全解析

1. 为什么批处理这件事值得单独拿出来聊

但凡写过几天 Python 脚本的人，迟早都会撞上同一个需求：手头有一堆外部命令要跑，可能是批量转码、批量压缩、批量调用某个命令行工具，也可能是把十几个脚本串成一条流水线。这时候你会发现，os.system()太糙，os.popen()又太局限，真正能扛事的还是subprocess模块。

subprocess是 Python 标准库里专门用来创建和管理子进程的模块，它替代了早期那一批又老又散的os.system、os.spawn*、os.popen等接口。它的核心价值在于：让你在 Python 里启动一个外部程序，并且能精确控制它的输入、输出、错误流，还能拿到它的返回码、超时状态，甚至实时读取它边跑边吐出来的日志。批处理场景下，这些能力一个都不能少。

这篇文章适合谁看？如果你已经会写基本的 Python 脚本，但对subprocess的印象还停留在“抄一句subprocess.run(cmd, shell=True)就完事”，那这篇就是写给你的。我会从批处理的整体设计思路讲起，把参数选型、并发控制、超时处理、日志采集、错误排查这些实战里真正会踩的坑一个个拆开，最后给出一套可以直接抄作业的批处理骨架。全程不玩虚的，都是我在实际项目里反复验证过的东西。

需要先说明一点：下面涉及的具体命令、路径、文件名都是示意性的，你替换成自己环境里的真实内容即可。批处理的本质逻辑是通用的，跟具体跑的是什么命令关系不大。

2. 批处理整体设计与 subprocess 选型思路

2.1 批处理到底在解决什么问题

先想清楚批处理这件事的本质。假设你手上有 500 个视频文件要统一转码，或者 300 个数据文件要跑同一个分析脚本，又或者 50 个服务要按顺序重启。这些任务的共同特征是：单个任务逻辑固定、重复次数多、彼此之间要么完全独立要么有明确依赖。

用 Python 做批处理，最大的好处是你能把“调度逻辑”和“被调用的外部程序”解耦。外部程序可以是 ffmpeg、可以是某个编译好的二进制、可以是另一个 Python 脚本，Python 这边只负责决定“跑哪些、跑几个、怎么跑、跑挂了怎么办”。这种解耦让整个流程变得可维护——换一个转码工具，你只需要改命令字符串，调度框架纹丝不动。

而subprocess就是这座桥。它负责把 Python 进程和外部程序进程连起来，让你既能发号施令，又能收结果。

2.2 为什么不用 os.system 和 os.popen

很多人第一反应是os.system，因为它简单，一行就能跑命令。但它在批处理场景下几乎是不可用的，原因有几个。

第一，os.system拿不到命令的输出。它只返回一个退出状态码，而且这个状态码在不同平台上含义还不一样，Windows 和 Linux 下拿到的值可能对不上。你想收集每个任务的执行日志？没门。

第二，os.system没法控制输入输出流。它默认把子进程的 stdout 和 stderr 直接打到当前终端，你没法把它们重定向到变量或者文件里做后续处理。

第三，os.system的错误处理极其粗糙。命令跑失败了，你只能靠返回码猜，拿不到任何错误信息。

os.popen稍微好一点，能拿到输出，但它只能单向通信（要么读要么写），而且返回的是一个文件对象，管理起来很别扭，超时控制、并发控制基本无从谈起。

subprocess把这些短板全补上了。它提供了run、Popen、call、check_output等一整套接口，覆盖从“跑一条命令拿结果”到“长期管理一个交互式进程”的所有场景。批处理里最常用的就是run（同步阻塞跑完拿结果）和Popen（异步启动，自己控制等待和读取）。

2.3 同步还是异步：run 与 Popen 的取舍

这是批处理设计里第一个关键决策。

subprocess.run()是同步的，调用它会一直阻塞到子进程结束，然后返回一个CompletedProcess对象，里面有returncode、stdout、stderr。它适合“一个接一个顺序执行”的场景，代码写起来最直观。

subprocess.Popen()是异步的，调用后立刻返回一个Popen对象，子进程在后台跑，你可以继续干别的事，需要的时候再通过wait()、poll()、communicate()去跟它交互。它适合“同时跑多个任务”或者“需要边跑边读输出”的场景。

批处理里这两者往往要混用。比如你要并发跑 8 个转码任务，那就得用Popen一次性启动 8 个，然后轮询它们的状态；但每个任务内部如果还有串行的小步骤，那些步骤用run就够了。

我的经验是：能用 run 就别用 Popen。run帮你处理了绝大多数边界情况（比如自动等待、自动收集输出），代码更短更不容易出错。只有当run满足不了需求——比如要并发、要实时读日志、要中途干预——才升级到Popen。

2.4 并发模型的选择：线程池还是进程池

批处理要提速，绕不开并发。但这里有个容易搞混的点：你要并发的是“外部命令”，不是“Python 计算”。

因为每个任务的实际工作是由外部程序（比如 ffmpeg）在自己的进程里完成的，Python 这边只是负责启动和等待，属于 IO 密集型操作。这种情况下用ThreadPoolExecutor线程池就够了，不需要ProcessPoolExecutor。线程在等待子进程结束的时候会释放 GIL，所以并发启动多个外部命令完全没问题。

用进程池反而更重，每个 Python 进程都要复制一份内存，启动开销大，而且进程间通信麻烦。除非你的批处理里还夹杂着大量纯 Python 的 CPU 计算，否则线程池是更优解。

并发数怎么定？不是越大越好。外部程序本身也要吃 CPU、内存、磁盘 IO。如果你同时跑 32 个 ffmpeg 转码，机器大概率直接卡死，每个任务都变慢，总时间反而更长。一般建议并发数设成 CPU 核心数，或者核心数的 1.5 倍。如果是 IO 密集型的命令（比如大量小文件拷贝），可以适当调高。这个值最好做成可配置参数，方便在不同机器上调整。

3. subprocess 核心参数逐个拆解与实操要点

3.1 args 的两种写法：字符串还是列表

subprocess的第一个参数args可以传字符串，也可以传列表。这两种写法差别很大，用错了会出各种诡异问题。

传列表是推荐做法，比如["ffmpeg", "-i", "input.mp4", "output.mp4"]。每个元素是一个独立的参数，Python 会直接把它们传给操作系统，不经过 shell 解析。这样最安全，参数里带空格、带特殊字符都不会出问题。

传字符串，比如"ffmpeg -i input.mp4 output.mp4"，只有在shell=True的时候才有意义。这时候字符串会交给系统的 shell 去解析，shell 会做变量替换、通配符展开、管道重定向等操作。

什么时候必须用字符串加shell=True？当你需要用到 shell 特性的时候，比如管道|、重定向>、通配符*、环境变量$HOME。比如"cat access.log | grep 500 | wc -l"这种，就只能走 shell。

但shell=True有安全风险。如果命令字符串里拼接了外部传入的参数，就可能被注入恶意命令。批处理里如果文件名来自不可信来源，千万别用shell=True拼字符串。正确做法是用列表传参，需要管道的话在 Python 里用多个Popen串起来，或者干脆用 Python 自己的字符串处理代替 grep、wc 这些。

提示：能用列表就别用字符串，能用 shell=False 就别开 shell=True。这是 subprocess 使用的第一安全原则。

3.2 stdout 和 stderr 的捕获方式

默认情况下，子进程的 stdout 和 stderr 会继承父进程的，也就是直接打到你的终端上。批处理里你通常想把它们抓下来，方便记录日志和排查问题。

subprocess.run()里用capture_output=True就能同时捕获 stdout 和 stderr，等价于分别设置stdout=subprocess.PIPE和stderr=subprocess.PIPE。跑完之后从返回对象的.stdout和.stderr里取，拿到的是 bytes，需要.decode()转成字符串。

这里有个大坑：PIPE 是有容量限制的。操作系统给管道分配的缓冲区通常只有几十 KB。如果你用Popen启动了子进程，子进程疯狂往 stdout 写数据，但你一直不去读，缓冲区写满之后子进程就会阻塞在写操作上，永远跑不完。这就是经典的“管道死锁”。

subprocess.run()不会有这个问题，因为它内部用communicate()帮你把输出读干净了。但如果你手动用Popen并且设了PIPE，就必须用communicate()或者自己开线程去读，不能傻等wait()。

批处理里如果子进程输出量很大（比如转码工具会打印大量进度信息），建议把 stdout 直接重定向到文件，而不是用 PIPE。用stdout=open("task.log", "w")这种方式，让操作系统直接写文件，既不会死锁，也不占 Python 内存。

3.3 超时控制：timeout 参数的正确用法

批处理最怕的就是某个任务卡死，把整条流水线拖住。subprocess.run()的timeout参数就是干这个的：超过指定秒数还没结束，就抛TimeoutExpired异常，并且自动杀掉子进程。

用法很直接：subprocess.run(cmd, timeout=300)表示最多等 300 秒。超时后异常里会带上已经捕获到的部分输出，方便你判断卡在哪一步。

但要注意，timeout杀掉的是直接启动的那个进程。如果这个进程自己又 fork 了子进程（比如 shell 脚本里启动的后台任务），那些孙进程可能不会被一起杀掉，变成孤儿进程继续占资源。要彻底清理，得用进程组的方式启动，超时后杀整个进程组。这个在 Linux 下通过start_new_session=True配合os.killpg实现，稍微复杂一点，后面实操部分会讲。

Popen没有timeout参数，得自己用wait(timeout=...)或者communicate(timeout=...)来控制，超时后手动调kill()。

3.4 check 参数与返回码处理

subprocess.run()默认不管子进程返回码是多少，都正常返回。如果你希望“命令失败就抛异常”，加上check=True。这时候返回码非零会抛CalledProcessError，异常对象里有returncode、cmd、stdout、stderr，信息很全。

批处理里我一般不用check=True，而是自己判断returncode。原因是批处理往往需要“某个任务失败后继续跑其他任务，最后统一汇总失败列表”，而不是一失败就整个崩掉。自己判断返回码更灵活，能记录下是哪个任务、什么命令、什么错误，最后一起处理。

返回码的约定：0 表示成功，非零表示失败。但不同程序对非零码的含义不一样，有的用 1 表示一般错误，有的用 2 表示参数错误。批处理里不用太纠结具体数值，只要非零就当失败记录，把 stderr 一起存下来供后续分析就够了。

3.5 环境变量与工作目录

env参数可以给子进程指定一套独立的环境变量。默认情况下子进程继承父进程的环境。如果你想让子进程用不同的 PATH、不同的语言设置，就传一个字典进去。注意传env会完全替换掉环境，不是追加，所以通常要先os.environ.copy()再改。

cwd参数指定子进程的工作目录。这个在批处理里很有用：很多命令行工具会根据当前目录找配置文件或者输出文件，你不想被父进程的目录影响，就显式指定cwd。

这两个参数配合起来，能让每个任务在干净、可控的环境里运行，避免任务之间互相干扰。

4. 批处理实操：从单任务到并发流水线

4.1 最小可用版本：顺序跑一批命令

先从最简单的开始。假设你有一个命令模板，要对一批输入文件依次执行。

import subprocess import logging logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") def run_one(cmd, timeout=600): try: result = subprocess.run( cmd, capture_output=True, timeout=timeout, text=True, ) if result.returncode == 0: logging.info("成功: %s", " ".join(cmd)) return True, result.stdout else: logging.error("失败(码%d): %s\n%s", result.returncode, " ".join(cmd), result.stderr) return False, result.stderr except subprocess.TimeoutExpired as e: logging.error("超时: %s", " ".join(cmd)) return False, "timeout" except FileNotFoundError: logging.error("命令不存在: %s", cmd[0]) return False, "not found" files = [f"input_{i}.dat" for i in range(1, 21)] ok, fail = 0, 0 for f in files: cmd = ["my_tool", "--input", f, "--output", f + ".out"] success, _ = run_one(cmd) if success: ok += 1 else: fail += 1 logging.info("完成: 成功 %d, 失败 %d", ok, fail)

这段代码有几个细节值得说。text=True让 stdout 和 stderr 自动按文本解码，省得你手动.decode()，默认用系统编码，也可以显式指定encoding="utf-8"。异常处理分了三种：超时、命令不存在、其他。FileNotFoundError是命令本身找不到时抛的，跟命令跑失败是两码事，分开处理更清晰。

顺序版本的好处是逻辑简单、资源占用低、日志顺序清晰。缺点是慢。20 个任务如果每个要 30 秒，那就是 10 分钟。任务一多就受不了。

4.2 并发版本：线程池 + Popen 组合

要提速，上线程池。这里我用ThreadPoolExecutor配合subprocess.run，因为run本身是阻塞的，放在线程里跑正好。

import subprocess import logging from concurrent.futures import ThreadPoolExecutor, as_completed logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") def run_one(cmd, timeout=600): try: result = subprocess.run(cmd, capture_output=True, timeout=timeout, text=True) return cmd, result.returncode, result.stdout, result.stderr except subprocess.TimeoutExpired: return cmd, -1, "", "timeout" except FileNotFoundError: return cmd, -2, "", "command not found" def batch_run(commands, max_workers=4, timeout=600): results = [] with ThreadPoolExecutor(max_workers=max_workers) as pool: future_map = { pool.submit(run_one, cmd, timeout): cmd for cmd in commands } for future in as_completed(future_map): cmd, code, out, err = future.result() results.append((cmd, code, out, err)) if code == 0: logging.info("成功: %s", " ".join(cmd)) else: logging.error("失败(码%d): %s | %s", code, " ".join(cmd), err.strip()[:200]) return results commands = [ ["my_tool", "--input", f"input_{i}.dat", "--output", f"input_{i}.dat.out"] for i in range(1, 21) ] results = batch_run(commands, max_workers=4) failed = [r for r in results if r[1] != 0] logging.info("总计 %d, 失败 %d", len(results), len(failed))

max_workers=4表示最多同时跑 4 个任务。as_completed让结果按完成顺序返回，而不是按提交顺序，这样日志能实时反映进度。每个任务的输出都单独收集，最后统一汇总。

这里有个细节：run_one里捕获了TimeoutExpired和FileNotFoundError，返回统一的四元组。这样上层处理逻辑不用关心异常，只看返回码就行。返回码用负数区分异常类型（-1 超时，-2 命令不存在），跟正常的非零返回码区分开。

4.3 实时读取输出：边跑边看日志

有些任务跑得久，你希望实时看到它的输出，而不是等它跑完才一次性拿到。这时候run就不行了，得用Popen配合逐行读取。

import subprocess import logging def run_streaming(cmd, timeout=600): proc = subprocess.Popen( cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1, ) lines = [] try: for line in proc.stdout: line = line.rstrip("\n") lines.append(line) logging.info("[%s] %s", cmd[0], line) proc.wait(timeout=timeout) except subprocess.TimeoutExpired: proc.kill() proc.wait() logging.error("超时被杀: %s", " ".join(cmd)) return -1, "\n".join(lines) return proc.returncode, "\n".join(lines)

关键点：stderr=subprocess.STDOUT把错误流合并到标准输出，这样你只需要读一个流，日志顺序也不会乱。bufsize=1配合text=True表示行缓冲，让子进程每输出一行你就能读到一行，而不是攒一大块才给你。

for line in proc.stdout这个循环会一直读到子进程关闭 stdout。如果子进程卡住不输出也不退出，这个循环就会一直等，所以超时控制得靠外层。这里我在wait上设了 timeout，但注意如果子进程一直不关 stdout，for循环本身就会阻塞，wait根本执行不到。更稳妥的做法是把读取放到单独线程里，主线程负责超时监控。这个后面排查部分会细说。

4.4 超时后彻底清理：进程组方案

前面提过，timeout只杀直接子进程，孙进程可能漏网。要彻底清理，得让子进程独立成一个进程组，超时后杀整个组。

import subprocess import os import signal import logging def run_with_group_kill(cmd, timeout=600): proc = subprocess.Popen( cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, start_new_session=True, ) try: out, err = proc.communicate(timeout=timeout) return proc.returncode, out, err except subprocess.TimeoutExpired: try: os.killpg(os.getpgid(proc.pid), signal.SIGTERM) proc.wait(timeout=10) except (ProcessLookupError, subprocess.TimeoutExpired): try: os.killpg(os.getpgid(proc.pid), signal.SIGKILL) except ProcessLookupError: pass return -1, "", "timeout and killed"

start_new_session=True让子进程成为新会话的组长，它和它派生的所有进程都属于同一个进程组。超时后先发SIGTERM给整个组，给程序一个优雅退出的机会，等 10 秒还不退就上SIGKILL强杀。这样能最大程度避免孤儿进程。

这套方案在 Linux 和 macOS 下有效。Windows 下进程组的概念不一样，得用CREATE_NEW_PROCESS_GROUP标志配合taskkill /T来杀进程树，写法不同，这里不展开。

4.5 完整批处理骨架：可直接抄作业

把上面的东西拼起来，就是一个能用的批处理框架。

import subprocess import os import signal import logging import time from concurrent.futures import ThreadPoolExecutor, as_completed logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s", handlers=[ logging.FileHandler("batch.log", encoding="utf-8"), logging.StreamHandler(), ], ) def run_task(cmd, timeout=600, cwd=None, env=None): start = time.time() proc = subprocess.Popen( cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, cwd=cwd, env=env, start_new_session=True, ) try: out, err = proc.communicate(timeout=timeout) elapsed = time.time() - start return { "cmd": cmd, "code": proc.returncode, "stdout": out, "stderr": err, "elapsed": elapsed, } except subprocess.TimeoutExpired: try: os.killpg(os.getpgid(proc.pid), signal.SIGTERM) proc.wait(timeout=10) except (ProcessLookupError, subprocess.TimeoutExpired): try: os.killpg(os.getpgid(proc.pid), signal.SIGKILL) except ProcessLookupError: pass return { "cmd": cmd, "code": -1, "stdout": "", "stderr": "timeout", "elapsed": time.time() - start, } def batch(commands, max_workers=4, timeout=600, cwd=None, env=None): results = [] with ThreadPoolExecutor(max_workers=max_workers) as pool: futures = { pool.submit(run_task, cmd, timeout, cwd, env): cmd for cmd in commands } for fut in as_completed(futures): r = fut.result() results.append(r) if r["code"] == 0: logging.info("OK %.1fs %s", r["elapsed"], " ".join(r["cmd"])) else: logging.error( "FAIL code=%d %.1fs %s | %s", r["code"], r["elapsed"], " ".join(r["cmd"]), r["stderr"].strip()[:300], ) return results if __name__ == "__main__": cmds = [ ["my_tool", "--input", f"data_{i}.bin", "--output", f"data_{i}.out"] for i in range(1, 51) ] res = batch(cmds, max_workers=4, timeout=300) ok = sum(1 for r in res if r["code"] == 0) logging.info("全部完成: 成功 %d / %d", ok, len(res))

这个骨架覆盖了：并发控制、超时、进程组清理、日志落盘、耗时统计、失败汇总。你可以直接改cmds的生成逻辑和max_workers，套到自己的场景里。

5. 常见问题与排查技巧实录

5.1 命令跑不起来：FileNotFoundError 排查

最常见的报错就是FileNotFoundError: [Errno 2] No such file or directory。这通常不是文件找不到，而是命令本身找不到。

原因一般是命令不在 PATH 里。比如你在终端能直接敲ffmpeg，但 Python 里跑就报错，很可能是 Python 进程的 PATH 跟你交互式 shell 的 PATH 不一样。解决办法有两个：一是用命令的绝对路径，比如/usr/local/bin/ffmpeg；二是显式传env，把 PATH 设对。

还有一种情况是命令名拼错了，或者大小写不对。Linux 下大小写敏感，FFmpeg和ffmpeg是两个东西。

排查方法很简单：在 Python 里打印os.environ.get("PATH")，再shutil.which("你的命令")看看能不能找到。找不到就说明 PATH 有问题。

5.2 中文乱码：编码问题处理

子进程输出的中文变成乱码，是编码不匹配导致的。text=True默认用系统编码解码，Windows 下可能是 GBK，Linux 下可能是 UTF-8。如果子进程实际输出的是 UTF-8，但系统编码是 GBK，就会乱码。

解决办法是显式指定encoding="utf-8"，并且加上errors="replace"防止个别非法字节导致整个解码失败抛异常。

result = subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8", errors="replace")

如果子进程输出的是二进制数据（比如图片、压缩包），千万别用text=True，直接拿 bytes 自己处理。

5.3 卡死不动：管道死锁的识别与解决

程序跑着跑着不动了，CPU 占用很低，也不报错，八成是管道死锁。

判断方法：用ps或者任务管理器看子进程状态，如果它处于S（睡眠）或者D（不可中断睡眠）状态，而且一直不结束，同时 Python 进程也在等，那就是死锁了。

根因就是前面说的：Popen设了PIPE，子进程写满了缓冲区，但父进程没在读。解决办法：要么用communicate()（它会同时读 stdout 和 stderr），要么把输出重定向到文件，要么自己开线程读。

subprocess.run()内部用了communicate()，所以不会有这个问题。这也是我推荐优先用run的原因之一。

5.4 返回码非零但没错误信息

有时候命令返回非零，但 stderr 是空的，你完全不知道哪里错了。这种情况通常是命令把错误信息打到了 stdout，而不是 stderr。有些工具就是这么不守规矩。

解决办法是把 stdout 和 stderr 都收集起来一起看。或者用stderr=subprocess.STDOUT把两者合并，这样不管它打到哪个流，你都能拿到。

还有一种可能是命令通过返回码传递信息，比如返回 1 表示“没有匹配项”而不是“出错”。这种得查具体工具的文档，不能一概而论。

5.5 并发数设太大导致系统卡顿

并发数不是越大越好。我见过有人设max_workers=64，结果机器直接卡到没法操作，所有任务都变慢，总耗时反而比串行还长。

原因是每个外部程序都要抢 CPU、内存、磁盘 IO。并发太高，操作系统频繁切换上下文，磁盘随机读写暴增，整体效率断崖式下降。

经验值：CPU 密集型命令（转码、压缩）设成核心数；IO 密集型命令（拷贝、网络请求）可以设成核心数的 2 到 4 倍。最好做成参数，实际跑的时候观察系统负载再调。

5.6 常见问题速查表

5.7 几个我踩过的坑

第一个坑：用shell=True拼路径，结果文件名里带空格，命令被拆成了两半。后来全部改成列表传参，再没出过这问题。

第二个坑：Popen设了PIPE但只读 stdout 不读 stderr，子进程往 stderr 写满之后卡死。这个坑很隐蔽，因为 stdout 看起来一切正常，就是任务不结束。后来养成习惯，要么两个都读，要么合并成一个流。

第三个坑：超时杀了主进程，但它启动的子进程还在后台跑，占着端口不放。后来加了start_new_session=True和killpg才彻底解决。

第四个坑：并发跑的时候日志交错在一起，根本分不清哪行是哪个任务的。后来在每个任务的输出前面加上任务标识，日志才可读。

6. 性能调优与进阶技巧

6.1 减少进程启动开销

每次subprocess调用都要 fork 一个新进程，这个开销在任务数量大、单个任务很短的时候会变得很明显。比如你要跑 10000 个只耗时 10 毫秒的小命令，光进程启动可能就占了大半时间。

优化思路：能合并的命令就合并。比如把 100 个文件一次性传给一个命令处理，而不是循环调用 100 次。很多命令行工具都支持批量输入，用好了能省掉大量进程启动开销。

如果实在没法合并，可以考虑用multiprocessing的进程池，或者干脆把外部命令的逻辑用 Python 重写。当然这取决于具体场景，不是所有命令都能重写。

6.2 输出重定向到文件 vs PIPE

前面提过，输出量大时用文件重定向比 PIPE 好。这里补充一点：文件重定向还有个好处是，即使 Python 进程崩了，子进程的输出也已经落盘了，不会丢。而 PIPE 里的数据在 Python 进程崩溃后就没了。

with open("task_output.log", "w", encoding="utf-8") as f: result = subprocess.run(cmd, stdout=f, stderr=subprocess.STDOUT, timeout=600)

这样 stdout 和 stderr 都写到同一个文件，顺序也保持。跑完之后需要的话再读文件分析。

6.3 用 shlex 安全地构造命令

如果你的命令参数来自用户输入或者配置文件，用shlex.quote()可以安全地转义特殊字符，防止注入。

import shlex filename = "my file; rm -rf /" safe = shlex.quote(filename) cmd = f"my_tool --input {safe}"

不过更推荐的做法还是用列表传参，从根本上避免 shell 解析。shlex主要是在你不得不拼字符串的时候用。

6.4 任务依赖与流水线编排

批处理不总是“一堆独立任务”。有时候任务之间有依赖，比如 B 必须在 A 成功之后才能跑。这种用简单的线程池就不够了，得引入依赖管理。

简单场景可以用拓扑排序把任务排成几批，每批内部并发，批与批之间串行。复杂场景可能需要引入专门的编排工具，但那已经超出subprocess的范畴了。

我的建议是：依赖关系尽量在 Python 层面用代码表达清楚，别塞进 shell 脚本里。Python 的条件判断、异常处理比 shell 强太多，维护起来也容易。

7. 一些个人体会

subprocess这个模块，表面上看就是“跑个命令”，但真到批处理场景里，坑是一个接一个。我自己的经验是，把下面这几件事做扎实，能省掉 80% 的麻烦。

第一，永远用列表传参，除非你明确需要 shell 特性。这一条能挡掉大部分安全和转义问题。

第二，永远设超时。没有超时的批处理就是定时炸弹，迟早有一个任务卡死把整条线拖垮。

第三，输出要么用communicate读干净，要么重定向到文件。别让 PIPE 里的数据没人管。

第四，并发数要保守。先设小一点，观察系统负载再往上调。宁可慢一点，也别把机器跑挂。

第五，日志要带任务标识和耗时。出问题的时候，这两样东西能帮你快速定位是哪个任务、卡了多久。

最后分享一个小技巧：批处理脚本最好支持“断点续跑”。也就是记录哪些任务已经成功，重跑的时候跳过它们。实现很简单，成功一个就往一个文件里追加一行任务标识，启动时先读这个文件过滤掉已完成的。这个功能在任务多、耗时长的时候特别有用，跑挂了不用从头再来。

这套东西我在好几个项目里反复用过，从几百个文件的批量处理到几十个服务的批量操作，基本都能扛住。核心逻辑就那些，剩下的就是根据具体场景微调参数。你要是刚开始用subprocess做批处理，建议先把第 4 节那个骨架跑通，再往上加自己的逻辑，比一上来就写复杂流程要稳得多。