如何在FastAPI中异步运行后台进程并获取返回数据？

你的核心问题是：subprocess.run()是同步阻塞操作，即便放在async def路由里，也会卡住FastAPI的事件循环，导致后续请求必须等待前一个完成。用threading没成功是因为没正确将阻塞操作从事件循环线程中剥离，下面是具体的修改方案：

方案：用线程池异步执行subprocess调用 #

我们可以用concurrent.futures.ThreadPoolExecutor把阻塞的Java程序调用放到独立线程中，让FastAPI的事件循环能同时处理多个请求，实现并行执行。

修改后的完整代码 #

from fastapi import FastAPI, status
from fastapi.responses import JSONResponse
import subprocess
from datetime import datetime
import asyncio
from concurrent.futures import ThreadPoolExecutor

app = FastAPI()

# 创建全局线程池，避免每次请求创建销毁线程
executor = ThreadPoolExecutor(max_workers=4)  # 根据需求调整最大线程数

def run_java_program():
    """封装Java程序调用，放到线程池执行"""
    try:
        msg = subprocess.run(
            ['java', '-cp', 'AsyncTest.jar', 'AsyncTest'],
            cwd='java_files',
            capture_output=True,
            check=True,
            text=True
        )
        return msg.stdout if msg.stdout else msg.stderr
    except subprocess.CalledProcessError as e:
        # 捕获Java程序执行错误（比如返回非0状态码）
        return e.stderr
    except Exception as e:
        return f"error running script: {str(e)}"

@app.get("/test")
async def test_java():
    print(f'Starting: {datetime.now()}')
    # 把阻塞操作提交到线程池，转为可await的对象
    results = await asyncio.wrap_future(executor.submit(run_java_program))
    
    print(f'Finishing: {datetime.now()}')
    
    if results.startswith("error running script"):
        return JSONResponse(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            content={'results': results}
        )
    
    return JSONResponse(
        status_code=status.HTTP_200_OK,
        content={'results': results}
    )

关键改动说明 #

• 封装阻塞操作：把subprocess.run()放到单独的同步函数run_java_program里，这个函数负责执行Java程序并返回结果/错误信息。
• 全局线程池：创建全局的ThreadPoolExecutor，复用线程资源，比每次请求创建更高效。
• 异步适配：用asyncio.wrap_future()把线程池返回的Future对象转为FastAPI事件循环可await的对象，这样事件循环在等待Java程序执行时，能处理其他请求。

这样修改后，间隔1秒点击两次按钮，两个Java进程会并行执行，总耗时约31秒，而非原来的60秒。

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