如何在FastAPI中使用指定Executor启动asyncio子进程

错误原因 #

你遇到的TypeError: cannot pickle 'coroutine' object错误，根源在于：

• asyncio.create_subprocess_shell是协程函数，调用它会返回协程对象
• loop.run_in_executor配合ProcessPoolExecutor使用时，需要将任务序列化（pickle）后传给子进程，但协程对象无法被pickle序列化，因此报错

正确做法 #

场景1：仅需异步执行外部命令（推荐） #

asyncio.create_subprocess_shell本身就是异步非阻塞的，它基于事件循环处理IO，不会阻塞FastAPI的事件循环，完全不需要额外的Executor。你原来的代码就是正确的：

proc = await asyncio.create_subprocess_shell(
    cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
)
# 逐行读取stdout示例
async for line in proc.stdout:
    print(line.decode().strip())

场景2：必须使用指定Executor（如CPU密集型子任务） #

如果确实需要通过ProcessPoolExecutor管理任务（比如执行CPU密集型的本地代码，而非单纯调用外部命令），需要将同步执行逻辑封装成普通函数，再交给run_in_executor：

from concurrent.futures import ProcessPoolExecutor
import subprocess
import asyncio
from fastapi import FastAPI

app = FastAPI()
# 初始化指定的Executor
pool = ProcessPoolExecutor(max_workers=10)

def run_command_sync(cmd):
    # 使用同步的subprocess API执行命令
    result = subprocess.run(
        cmd, 
        shell=True, 
        capture_output=True, 
        text=True
    )
    return result.stdout, result.stderr

@app.get("/execute")
async def execute_command():
    cmd = "your_command_here"
    loop = asyncio.get_running_loop()
    # 在指定Executor中执行同步任务
    stdout, stderr = await loop.run_in_executor(pool, run_command_sync, cmd)
    return {"stdout": stdout, "stderr": stderr}

关键说明 #

• asyncio的subprocess API专为异步IO设计，调用外部命令时优先使用，无需额外线程/进程池
• ProcessPoolExecutor更适合处理CPU密集型的本地计算任务，而非IO密集型的外部命令调用

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