Python异步进程回调失败时的正确跨平台终止方案

问题核心：当用户回调抛出异常时，异步循环提前中断，后续的process.communicate()未执行，导致子进程的stdout/stderr管道及底层transport资源未被正确清理，从而触发“Resource warning: unclosed transport”警告。直接调用terminate()仅终止进程，无法完成资源释放。

正确处理流程（跨Windows/Linux，Python3.11兼容） #

捕获回调异常后，需先终止进程，再通过communicate()读取剩余输出并关闭管道（这是asyncio提供的公开API，无需访问私有变量），最后重新抛出异常。完整代码如下：

import asyncio
import subprocess

async def run_subprocess(subprocess_args, user_provided_callback):
    # 启动异步子进程
    process = await asyncio.subprocess.create_subprocess_exe(
        subprocess_args,
        stderr=asyncio.subprocess.PIPE,
        stdout=asyncio.subprocess.PIPE
    )

    try:
        # 异步读取stdout并调用回调
        async for line in process.stdout:
            user_provided_callback(line)
    except Exception as err:
        # 终止子进程
        process.terminate()
        # 读取剩余输出并关闭管道，彻底释放资源
        _, stderr = await process.communicate()
        # 可选：将stderr信息附加到异常中便于调试
        raise Exception(f"回调执行失败: {err}, 子进程错误输出: {stderr.decode()}") from err

    # 正常流程下等待进程终止并检查返回码
    _, stderr = await process.communicate()
    if process.returncode != 0:
        raise Exception(f"子进程执行失败，错误输出: {stderr.decode()}")

关键说明 #

1. communicate()的作用：该方法会自动读取子进程剩余的stdout/stderr输出，并关闭所有关联的管道和底层transport，是asyncio官方推荐的资源清理方式，完全跨平台兼容。
2. 异常处理顺序：先终止进程再调用communicate()，既保证子进程停止运行，又确保所有资源被正确回收，彻底消除未关闭transport的警告。
3. 异常链保留：使用raise ... from err保留原始异常栈，便于定位回调出错的根源。

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