使用 Flask 构建流式返回服务

发布于:2025-02-14 ⋅ 阅读:(11) ⋅ 点赞:(0)

使用 Flask 构建流式返回服务是一个很常见的应用场景,特别是在需要逐步传输大数据或进行长时间操作的场景下(比如下载大文件、实时日志等)。Flask 中可以通过 Response 对象来实现流式响应。以下是一个简单的例子,展示了如何在 Flask 中构建一个流式返回的服务。

示例:构建一个流式返回的 Flask 服务

1. 创建 Flask 应用并实现流式返回功能
from flask import Flask, Response
import time

app = Flask(__name__)

# 流式返回的生成器
def generate_data():
    for i in range(10):
        # 模拟处理延时,例如从数据库获取数据或计算等
        time.sleep(1)
        yield f"data chunk {i}\n"

# 路由:返回流式数据
@app.route('/stream')
def stream():
    return Response(generate_data(), content_type='text/plain;charset=utf-8')

if __name__ == '__main__':
    app.run(debug=True)

解释:

  • generate_data() 是一个生成器函数,它逐个产生数据块,并在每个块之间模拟了1秒的延时(这通常代表一些计算、网络请求或数据库查询等操作)。
  • Response(generate_data()) 用于创建一个响应对象,它将逐步从生成器中获取数据并发送给客户端。
  • content_type='text/plain;charset=utf-8' 用于指定返回的内容类型,你可以根据实际需要调整。
2. 启动 Flask 服务

在命令行中运行上述代码,启动 Flask 应用:

python app.py

默认情况下,Flask 会在 http://127.0.0.1:5000/ 启动服务。

3. 使用流式服务

你可以使用 requests 库来请求这个流式返回的数据。以下是客户端代码:

import requests

url = 'http://127.0.0.1:5000/stream'

response = requests.get(url, stream=True)

# 逐行打印返回的内容
for line in response.iter_lines(decode_unicode=True):
    if line:
        print(line)

关键点:

  • stream=True 用于告诉 requests 客户端逐步接收响应内容。
  • iter_lines() 用于逐行读取流式数据。
4. 流式返回的其他应用

流式返回在以下场景中非常有用:

  • 实时日志监控:你可以使用流式响应实时输出应用的日志。
  • 大文件下载:你可以使用流式响应按块返回文件内容,避免一次性加载大文件到内存中。
  • 长时间计算任务:比如,实时处理并返回处理结果。

希望这个示例能帮助你理解如何在 Flask 中实现流式返回服务!你有任何问题或者需要进一步的例子吗?