FastAPI 流式响应入门教程：从零开始学习 StreamingResponse

欢迎来到FastAPI的世界！如果你有Python基础，但没做过Web开发，别担心，本教程会像学做一道新菜一样，带你一步步上手。今天，我们重点学习流式响应（StreamingResponse），这就像是水管里的水，而不是一次性倒满杯子的水。准备好了吗？让我们开始吧！

1. 准备你的厨房：安装必备工具

首先，确保你有Python（建议3.7+）。然后，用pip安装FastAPI和uvicorn（一个服务器）。打开终端或命令提示符，运行：

pip install fastapi uvicorn

这就好比买了厨具，一切准备就绪！

2. 什么是流式响应？一个生活化类比

想象一下：

• 普通响应：就像用一个大水桶装水，一次装满所有数据再返回。适合小数据，但如果数据太大或需要实时更新，就会很慢。
• 流式响应：就像一根水管，水一点点流出，可以持续传输数据。适合实时数据（如股票价格）或大文件（如视频流）。

在FastAPI中，StreamingResponse 就是一个这样的水管，让你一次返回一点数据，而不是等到所有数据都准备好。

3. 你的第一个流式API：送上一串祝福

让我们创建一个简单的FastAPI应用，返回一个逐步发送的消息流。这就像在水管中一点一点地流出祝福的话语。

步骤1：创建Python文件

新建一个文件，比如 streaming_demo.py，并输入以下代码：

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import time

# 创建FastAPI应用
app = FastAPI()

# 定义一个生成器函数，模拟流式数据
def message_stream():
    messages = ["你好！", "欢迎来到FastAPI教程。", "这是流式响应示例。", "再见！"]
    for message in messages:
        yield f"{message}\n"  # 每次yield一小段数据
        time.sleep(1)  # 模拟延迟，让输出更清晰

@app.get("/stream")
async def stream_messages():
    """
    流式端点：返回一个实时消息流
    """
    return StreamingResponse(message_stream(), media_type="text/plain")

# 要运行，在终端输入：uvicorn streaming_demo:app --reload

步骤2：运行并查看结果

在终端中，进入文件所在目录，运行：

uvicorn streaming_demo:app --reload

然后，打开浏览器访问 http://127.0.0.1:8000/stream。你会看到消息一个一个地出现，每秒钟一条，就像水流一样！这就是你的第一个“成功时刻”——你已经创建了一个流式API！

4. 深入StreamingResponse：更多示例

现在，我们来玩点更酷的。FastAPI的StreamingResponse可以处理各种数据类型。

示例1：流式JSON响应

假设我们要实时发送一个数字序列，比如1到10，每个数字作为一个JSON对象。修改代码：

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import json
import time

app = FastAPI()

def number_stream():
    for i in range(1, 11):
        yield json.dumps({"number": i}) + "\n"  # 将数据转为JSON格式
        time.sleep(0.5)

@app.get("/numbers")
async def stream_numbers():
    return StreamingResponse(number_stream(), media_type="application/json")

运行后访问 http://127.0.0.1:8000/numbers，你会看到数字一个接一个地输出在JSON格式中。试试看！

示例2：模拟实时数据流（比如传感器读数）

让我们模拟一个温度传感器，每隔2秒发送一次温度数据。这是一个更贴近生活的例子。

import random

def sensor_stream():
    while True:  # 无限循环，模拟持续数据流
        temperature = random.uniform(20.0, 30.0)  # 生成随机温度
        yield f"温度: {temperature:.2f}°C\n"
        time.sleep(2)

@app.get("/sensor")
async def stream_sensor():
    return StreamingResponse(sensor_stream(), media_type="text/plain")

访问 http://127.0.0.1:8000/sensor，你会看到温度数据不断更新。这就是流式响应的魅力——可以实时推送数据！

5. 实践挑战：自己动手，丰衣足食

现在轮到你试试了！创建一个新端点，比如 /greetings，让它流式返回你的名字和一条欢迎消息，每次延迟1秒。代码提示：

• 定义一个生成器函数，yield出字符串。
• 使用 StreamingResponse 返回。

完成后，运行应用并访问这个端点，享受自己创造的“成功时刻”！

6. 总结与下一步

恭喜！你已经学会了如何在FastAPI中使用StreamingResponse来创建流式响应。记住：

• 流式响应适合处理实时或大量数据。
• StreamingResponse 是FastAPI提供的一个方便工具，可以返回任何可迭代的数据流。
• 通过动手实践，你已经建立了信心，可以进一步探索FastAPI的其他功能，比如文件上传、WebSocket等。

继续学习，快速API开发之路还很长。推荐查看FastAPI官方文档，那里有更多高级示例。祝你编码愉快！