快捷導(dǎo)航

Python異步調(diào)用外部命令的踩坑實(shí)錄

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

文章總結(jié)了在重構(gòu)數(shù)據(jù)處理服務(wù)時(shí)遇到的5個(gè)異步子進(jìn)程相關(guān)問題及解決方法,文中的示例代碼講解詳細(xì),感興趣的小伙伴可以跟隨小編一起學(xué)習(xí)一下

坑1：在async函數(shù)里直接用 subprocess.run() 阻塞整個(gè)事件循環(huán)
坑2：stdout/stderr管道沒消費(fèi)導(dǎo)致死鎖
坑3：大量并發(fā)子進(jìn)程耗盡文件描述符
坑4：子進(jìn)程超時(shí)與僵死處理
坑5：Windows上的兼容性地獄
總結(jié)

最近在重構(gòu)一個(gè)數(shù)據(jù)處理服務(wù)，需要并發(fā)調(diào)用十幾個(gè)外部命令行工具（ffmpeg、wkhtmltopdf之類）。本來(lái)以為把 subprocess.run() 換成 asyncio.create_subprocess_exec() 就完事了，結(jié)果踩了一串坑，分享給同樣在折騰異步子進(jìn)程的同學(xué)。

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

這是最常犯的錯(cuò)。很多人知道async函數(shù)，但習(xí)慣了同步寫法：

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

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

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

async def process_video(path):
    # ? 異步等待子進(jìn)程
    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管道沒消費(fèi)導(dǎo)致死鎖

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

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

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

解決方法：始終用 communicate() 同時(shí)讀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()  # ? 同時(shí)消費(fèi)兩個(gè)管道
    return proc.returncode, stdout, stderr

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

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

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

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

解決方案：

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

# 2. 用Semaphore控制并發(fā)數(shù)
sem = asyncio.Semaphore(10)  # 最多10個(gè)并發(fā)子進(jìn)程

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. 或者臨時(shí)提高限制（需要權(quán)限）
# resource.setrlimit(resource.RLIMIT_NOFILE, (65536, 65536))

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

坑4：子進(jìn)程超時(shí)與僵死處理

有些命令行工具偶爾會(huì)卡死（說(shuō)的就是你，wkhtmltopdf）。communicate() 本身沒有超時(shí)參數(shù)（Python 3.11之前），直接await可能永遠(yuǎn)等不回來(lái)：

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

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

注意兩點(diǎn)：

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

import os
import signal

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

# 超時(shí)后殺整個(gè)進(jìn)程組
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上的兼容性地獄

如果你的服務(wù)需要跨平臺(tái)，Windows是一堆坑的集合：

create_subprocess_exec 在Windows上不支持 preexec_fn 參數(shù)（Windows沒有進(jìn)程組概念）
殺進(jìn)程要用 proc.terminate() 而不是發(fā)信號(hào)
路徑中的反斜杠和空格需要特殊處理
編碼問題：stdout默認(rèn)是系統(tǒng)編碼（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

總結(jié)

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

異步子進(jìn)程看起來(lái)簡(jiǎn)單，實(shí)際上涉及操作系統(tǒng)管道、進(jìn)程管理、信號(hào)處理等底層細(xì)節(jié)。踩完這些坑之后，我對(duì)"異步"這個(gè)概念理解深了不少——它不只是把def改成async def，而是要真正理解你的代碼在事件循環(huán)里是怎么調(diào)度的。

以上都是實(shí)際項(xiàng)目中遇到的問題，希望幫你少走彎路。

到此這篇關(guān)于Python異步調(diào)用外部命令的踩坑實(shí)錄的文章就介紹到這了,更多相關(guān)Python異步調(diào)用外部命令內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: