当前位置：首页 > news >正文

FastAPI中的流式响应:实现实时数据传输

news 2025/7/10 6:39:19

在Web应用程序开发中,有时我们需要处理大量数据或长时间运行的操作。在这些情况下,传统的一次性响应可能会导致客户端长时间等待,甚至超时。这就是流式响应(Streaming Response)发挥作用的地方。

为什么使用流式响应?

流式响应允许服务器在生成数据时逐步将其发送给客户端,而不是等待所有数据准备就绪后再一次性发送。这种方法有几个优点:

减少首字节时间(TTFB):客户端可以更快地开始接收数据。
更好的用户体验:用户可以看到部分结果,而不是长时间等待。
内存效率:服务器不需要在内存中保存整个响应。
适用于大文件传输:可以流式传输大文件,而不会耗尽服务器内存。

然而,流式响应也有一些缺点:

复杂性增加:实现和管理流式响应可能比简单的一次性响应更复杂。
错误处理:一旦开始流式传输,就难以处理中途出现的错误。
兼容性:某些代理服务器或客户端可能不完全支持流式响应。

FastAPI中实现流式响应

FastAPI提供了StreamingResponse类来实现流式响应。让我们通过一个例子来看看如何使用它:

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import timeapp = FastAPI()async def number_generator():for i in range(10):yield f"数字 {i}\n"time.sleep(1)  # 模拟耗时操作@app.get("/stream")
async def stream_numbers():return StreamingResponse(number_generator(), media_type="text/plain")

让我们来解析这段代码:

我们导入了必要的模块,包括FastAPI和StreamingResponse。
定义了一个异步生成器函数number_generator():
- 这个函数使用yield关键字生成数字。
- time.sleep(1)模拟了一个耗时操作,在实际应用中可能是数据库查询或复杂计算。
在路由处理函数stream_numbers()中:
- 我们返回一个StreamingResponse对象。
- 第一个参数是我们的生成器函数。
- media_type="text/plain"指定了响应的内容类型。

当客户端访问/stream端点时,服务器会开始流式传输数字,每秒发送一个数字,总共发送10个数字。

实际应用示例:流式文件下载

让我们看一个更实用的例子 - 流式传输大文件:

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import osapp = FastAPI()async def file_streamer(path):with open(path, mode="rb") as file_like:yield from file_like@app.get("/download/{filename}")
async def download_file(filename: str):file_path = os.path.join("files", filename)if not os.path.exists(file_path):return {"error": "文件不存在"}return StreamingResponse(file_streamer(file_path), media_type="application/octet-stream",headers={"Content-Disposition": f"attachment; filename={filename}"})

这个例子中:

file_streamer函数是一个异步生成器,它以二进制模式打开文件并逐块yield文件内容。
在download_file路由处理函数中:
- 我们首先检查文件是否存在。
- 然后返回一个StreamingResponse,使用file_streamer作为内容源。
- media_type="application/octet-stream"表示这是一个二进制文件。
- 我们添加了一个Content-Disposition头,告诉浏览器这是一个需要下载的附件。