Python異步調用外部命令的踩坑實錄

更新時間：2026年05月06日 08:29:11 作者：用戶396269106003

文章總結了在重構數據處理服務時遇到的5個異步子進程相關問題及解決方法,文中的示例代碼講解詳細,感興趣的小伙伴可以跟隨小編一起學習一下

坑1：在async函數里直接用 subprocess.run() 阻塞整個事件循環(huán)

這是最常犯的錯。很多人知道async函數，但習慣了同步寫法：

async def process_video(path):
    # ? 這會阻塞整個事件循環(huán)！
    result = subprocess.run(["ffmpeg", "-i", path, "output.mp4"], capture_output=True)
    return result.stdout

subprocess.run() 是同步阻塞調用。在async函數里直接用它，整個事件循環(huán)都會卡住，其他協程全部停擺。如果你的FastAPI接口里這么寫，一個請求就能把服務凍住。

正確做法是用 asyncio.create_subprocess_exec()：

async def process_video(path):
    # ? 異步等待子進程
    proc = await asyncio.create_subprocess_exec(
        "ffmpeg", "-i", path, "output.mp4",
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE
    )
    stdout, stderr = await proc.communicate()
    return stdout

坑2：stdout/stderr管道沒消費導致死鎖

這個坑極其隱蔽。當你創(chuàng)建子進程并設置了 stdout=PIPE，但忘記讀取輸出時：

async def run_tool(cmd):
    proc = await asyncio.create_subprocess_exec(*cmd, stdout=asyncio.subprocess.PIPE)
    # ? 如果子進程輸出了大量數據填滿管道緩沖區(qū)（通常64KB），
    # 子進程會阻塞在write()上，你的await也永遠不會返回
    await proc.wait()  # 死鎖！
    return proc.returncode

操作系統管道緩沖區(qū)有限，子進程往stdout寫滿了就卡住，等你來讀。但你只在 wait()，不去讀，雙方互相等——死鎖。

解決方法：始終用 communicate() 同時讀stdout和stderr：

async def run_tool(cmd):
    proc = await asyncio.create_subprocess_exec(
        *cmd,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE
    )
    stdout, stderr = await proc.communicate()  # ? 同時消費兩個管道
    return proc.returncode, stdout, stderr

如果確實不需要輸出，重定向到DEVNULL：

proc = await asyncio.create_subprocess_exec(
    *cmd,
    stdout=asyncio.subprocess.DEVNULL,
    stderr=asyncio.subprocess.DEVNULL
)

坑3：大量并發(fā)子進程耗盡文件描述符

每個子進程至少占3個fd（stdin/stdout/stderr的管道），加上communicate的緩沖區(qū)。我一開始并發(fā)起了50個子進程，直接 OSError: [Errno 24] Too many open files。

解決方案：

# 1. 查看當前限制
import resource
print(resource.getrlimit(resource.RLIMIT_NOFILE))  # 通常1024

# 2. 用Semaphore控制并發(fā)數
sem = asyncio.Semaphore(10)  # 最多10個并發(fā)子進程

async def run_with_limit(cmd):
    async with sem:
        proc = await asyncio.create_subprocess_exec(
            *cmd,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE
        )
        stdout, stderr = await proc.communicate()
        return proc.returncode, stdout

# 3. 或者臨時提高限制（需要權限）
# resource.setrlimit(resource.RLIMIT_NOFILE, (65536, 65536))

Semaphore是最靠譜的方式，既控制fd消耗，也避免把CPU打滿。

坑4：子進程超時與僵死處理

有些命令行工具偶爾會卡死（說的就是你，wkhtmltopdf）。communicate() 本身沒有超時參數（Python 3.11之前），直接await可能永遠等不回來：

# ? 可能永遠卡住
stdout, stderr = await proc.communicate()

# ? 用wait_for加超時
try:
    stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=30.0)
except asyncio.TimeoutError:
    proc.kill()  # 發(fā)SIGKILL
    await proc.wait()  # 等待進程回收，避免僵尸進程
    raise

注意兩點：

kill() 之后一定要 wait()，否則子進程變成僵尸進程占用PID
kill() 發(fā)SIGKILL是強制終止，如果子進程有子子進程，它們可能變成孤兒進程。更干凈的做法是殺進程組：

import os
import signal

# 創(chuàng)建子進程時指定新的進程組
proc = await asyncio.create_subprocess_exec(
    *cmd,
    stdout=asyncio.subprocess.PIPE,
    stderr=asyncio.subprocess.PIPE,
    preexec_fn=os.setsid  # 新進程組
)

# 超時后殺整個進程組
try:
    stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=30.0)
except asyncio.TimeoutError:
    os.killpg(os.getpgid(proc.pid), signal.SIGKILL)
    await proc.wait()

坑5：Windows上的兼容性地獄

如果你的服務需要跨平臺，Windows是一堆坑的集合：

create_subprocess_exec 在Windows上不支持 preexec_fn 參數（Windows沒有進程組概念）
殺進程要用 proc.terminate() 而不是發(fā)信號
路徑中的反斜杠和空格需要特殊處理
編碼問題：stdout默認是系統編碼（GBK），不是UTF-8

import sys

async def run_cross_platform(cmd):
    kwargs = {
        "stdout": asyncio.subprocess.PIPE,
        "stderr": asyncio.subprocess.PIPE,
    }
    
    if sys.platform != "win32":
        kwargs["preexec_fn"] = os.setsid
    
    proc = await asyncio.create_subprocess_exec(*cmd, **kwargs)
    
    try:
        stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=30)
    except asyncio.TimeoutError:
        if sys.platform == "win32":
            proc.terminate()
        else:
            os.killpg(os.getpgid(proc.pid), signal.SIGKILL)
        await proc.wait()
        raise
    
    # Windows編碼處理
    if sys.platform == "win32":
        stdout = stdout.decode("gbk", errors="replace")
        stderr = stderr.decode("gbk", errors="replace")
    
    return stdout, stderr

總結

坑	現象	解法
同步subprocess阻塞	事件循環(huán)卡死	用create_subprocess_exec
管道未消費	死鎖	communicate()或DEVNULL
fd耗盡	Too many open files	Semaphore控制并發(fā)
子進程僵死	永久掛起	wait_for超時+kill+wait
Windows兼容	各種報錯	條件分支+terminate