Python使用asyncio串联两个子进程并实现中断的问题

一、正确使用asyncio.create_subprocess_exec实现进程间管道（解决"Bad file descriptor"错误） #

用create_subprocess_exec时，需手动创建匿名管道，将第一个进程的stdout绑定到管道写端，第二个进程的stdin绑定到管道读端，同时关闭父进程中不需要的文件描述符，避免泄漏。

示例代码：

import asyncio
import os

async def run_piped_processes():
    # 创建匿名管道
    read_fd, write_fd = os.pipe()

    try:
        # 启动第一个进程：输出视频帧到管道写端
        proc1 = await asyncio.create_subprocess_exec(
            "your_frame_generator_cmd",  # 替换为你的帧生成命令
            stdout=write_fd,
            pass_fds=(write_fd,)  # 显式传递管道写端给子进程
        )

        # 启动第二个进程：ffmpeg从管道读端读取数据
        proc2 = await asyncio.create_subprocess_exec(
            "ffmpeg", "-i", "-", "output.mp4",  # 示例ffmpeg命令，从stdin读取输入
            stdin=read_fd,
            pass_fds=(read_fd,)  # 显式传递管道读端给子进程
        )

        # 父进程关闭管道两端（子进程已继承，父进程无需保留）
        os.close(read_fd)
        os.close(write_fd)

        # 等待进程运行，处理中断场景
        try:
            await asyncio.gather(proc1.wait(), proc2.wait())
        except asyncio.CancelledError:
            proc1.terminate()
            proc2.terminate()
            await proc1.wait()
            await proc2.wait()
    finally:
        # 确保管道描述符被关闭，避免资源泄漏
        try:
            os.close(read_fd)
        except OSError:
            pass
        try:
            os.close(write_fd)
        except OSError:
            pass

asyncio.run(run_piped_processes())

核心注意事项：

• 父进程必须关闭管道两端，否则子进程无法检测到管道关闭（比如第一个进程结束后，父进程仍持有写端，第二个进程的stdin不会收到EOF）。
• 通过pass_fds参数显式传递管道描述符，确保子进程能正常访问。

二、若必须使用asyncio.create_subprocess_shell，如何终止进程 #

create_subprocess_shell启动的是shell进程（如bash），你的两个业务进程是shell的子进程。直接终止shell进程不会自动转发信号给子进程，导致子进程继续运行。解决办法是终止整个进程组：

方法1：发送信号给进程组 #

import asyncio
import os
import signal

async def run_shell_process():
    cmd = "your_frame_generator_cmd | ffmpeg -i - output.mp4"  # 你的shell管道命令
    proc = await asyncio.create_subprocess_shell(cmd)

    try:
        await proc.wait()
    except asyncio.CancelledError:
        # 获取shell进程的进程组ID，发送SIGTERM给整个组
        os.killpg(os.getpgid(proc.pid), signal.SIGTERM)
        await proc.wait()

asyncio.run(run_shell_process())

方法2：强制开启进程组（针对部分默认不创建进程组的shell） #

在shell命令开头添加set -m，强制启用进程组，再按方法1发送信号：

cmd = "set -m; your_frame_generator_cmd | ffmpeg -i - output.mp4"

核心注意事项：

• os.killpg会将信号发送给整个进程组，确保shell和它的所有子进程都能收到终止信号，避免子进程变成孤儿进程继续运行。

内容的提问来源于stack exchange，提问作者Joald