### 实现FastAPI中的流式输出与多线程调用 在构建高性能Web应用程序时，能够实时推送数据给客户端是非常重要的特性之一。为了满足这一需求，在FastAPI框架下可以利用`EventSourceResponse`对象配合SSE (Server-Sent Events)协议完成服务器到浏览器端的数据流传输[^1]。 #### 使用`EventSourceResponse`实现流式输出 下面是一个简单的例子展示如何创建一个支持事件源响应的接口： ```python from fastapi import FastAPI, Request from sse_starlette.sse import EventSourceResponse import time app = FastAPI() @app.get('/stream') async def message_stream(request: Request): async def event_generator(): while True: if await request.is_disconnected(): break yield { "event": "new_message", "id": f"{int(time.time())}", "retry": "1000", # 客户端重连间隔时间(ms) "data": str(time.strftime('%X')) } await asyncio.sleep(1) return EventSourceResponse(event_generator()) ``` 此代码片段定义了一个GET路由 `/stream` ，它会每隔一秒向连接它的HTTP客户端发送当前的时间戳作为消息体的一部分。如果检测到客户端断开，则停止广播新消息并结束该次通信过程。 #### 结合多线程处理请求 为了让服务具备更好的扩展性和效率，可以通过引入Python标准库中的`concurrent.futures.ThreadPoolExecutor`来进行异步任务调度，从而允许后台执行耗时操作而不阻塞主线程的工作循环。以下是改进后的版本，展示了怎样安全地管理并发访问： ```python from concurrent.futures import ThreadPoolExecutor from contextlib import asynccontextmanager from typing import AsyncGenerator executor = ThreadPoolExecutor(max_workers=8) @asynccontextmanager async def managed_executor() -> AsyncGenerator[ThreadPoolExecutor]: try: yield executor finally: pass # 可在此处加入清理逻辑 @app.on_event('shutdown') def shutdown_event(): executor.shutdown(wait=True) @app.post("/process_video") async def process_video(file: UploadFile = File(...)): future = executor.submit(_handle_file_upload, file=file) result = await get_or_wait_for_result(future) return {"status": "success"} def _handle_file_upload(*args, **kwargs): ... async def get_or_wait_for_result(future): ... ``` 上述示例中，通过装饰器`@asynccontextmanager`封装了资源获取与释放的过程；而`managed_executor()`函数则提供了一种优雅的方式来确保在线程池关闭之前所有的子线程都已经完成了它们的任务。此外还注册了一个应用级别的钩子用于监听程序终止信号以便及时回收系统资源。