### FastAPI 中实现流式响应 为了实现在 FastAPI 应用程序中的流式输出，可以利用 `StreamingResponse` 类来创建一个持续发送数据给客户端的接口。这种方式非常适合用于长时间运行的操作或是实时更新的数据传输场景。 ```python from fastapi import FastAPI from fastapi.responses import StreamingResponse import time app = FastAPI() async def generate_data(): for i in range(10): yield f"data-{i}\n" await asyncio.sleep(1) @app.get("/streaming/") async def stream_response(): return StreamingResponse(generate_data(), media_type="text/plain") ``` 上述代码展示了如何定义一个异步函数 `generate_data()` 来逐步生成并返回数据片段，在每次迭代间暂停一秒模拟延迟效果；而 `/streaming/` 路由则通过 `StreamingResponse` 将这些数据作为连续流传递给客户端[^1]。 ### 多线程处理方式 当涉及到 CPU 密集型任务或多 I/O 操作时，采用多线程技术可以在不阻塞主线程的情况下提高整体性能。下面的例子说明了怎样在一个 FastAPI 请求处理器内启动新线程来进行后台作业： ```python from concurrent.futures import ThreadPoolExecutor from threading import Thread from typing import Any from fastapi import FastAPI, BackgroundTasks executor = ThreadPoolExecutor(max_workers=8) # 创建线程池管理器实例 def background_task(task_id: int) -> None: """执行耗时较长的任务""" print(f"Task {task_id}: Starting...") time.sleep(5) print(f"Task {task_id}: Finished.") async def run_in_threadpool(func, *args, **kwargs) -> Any: loop = asyncio.get_event_loop() result = await loop.run_in_executor(executor, func, *args, **kwargs) return result @app.post("/start-task/{task_id}") async def start_background_task(task_id: int, bg_tasks: BackgroundTasks): thread = Thread(target=lambda: executor.submit(background_task, task_id)) thread.start() bg_tasks.add_task(thread.join) return {"message": "Background task started."} ``` 这里引入了一个名为 `ThreadPoolExecutor` 的工具来自标准库 `concurrent.futures` ，它允许开发者轻松地管理和调度多个工作线程。每当收到 POST 请求访问路径 `/start-task/<id>` 时，就会触发一个新的子线程去执行指定 ID 对应的任务背景逻辑[^2]。