Python异步调用外部命令的踩坑实录

async def process_video(path):
    # ❌ 这会阻塞整个事件循环！
    result = subprocess.run(["ffmpeg", "-i", path, "output.mp4"], capture_output=True)
    return result.stdout

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

async def run_tool(cmd):
    proc = await asyncio.create_subprocess_exec(*cmd, stdout=asyncio.subprocess.PIPE)
    # ❌ 如果子进程输出了大量数据填满管道缓冲区（通常64KB），
    # 子进程会阻塞在write()上，你的await也永远不会返回
    await proc.wait()  # 死锁！
    return proc.returncode

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

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

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

# 2. 用Semaphore控制并发数
sem = asyncio.Semaphore(10)  # 最多10个并发子进程

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))

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

# ✅ 用wait_for加超时
try:
    stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=30.0)
except asyncio.TimeoutError:
    proc.kill()  # 发SIGKILL
    await proc.wait()  # 等待进程回收，避免僵尸进程
    raise

import os
import signal

# 创建子进程时指定新的进程组
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()

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