首先使用生成器函數(shù)配合Response類實(shí)現(xiàn)流式響應(yīng)，1. 定義生成器函數(shù)yield分塊數(shù)據(jù)；2. 通過(guò)Response(generator, mimetype=...)返回流；3. 根據(jù)場(chǎng)景選擇mimetype類型如text/plain、text/event-stream或text/csv；4. 前端可通過(guò)EventSource接收SSE格式消息；5. 大文件導(dǎo)出時(shí)結(jié)合StringIO逐行生成避免內(nèi)存溢出；6. 部署時(shí)需關(guān)閉Gunicorn或Nginx的緩衝機(jī)制以確保實(shí)時(shí)性；7. 注意在生成器中捕獲異常防止流中斷；該方法適用於實(shí)時(shí)日誌、AI逐字輸出、CSV導(dǎo)出等場(chǎng)景，能有效提升響應(yīng)效率和用戶體驗(yàn)。

如果你在用Flask 做Web 開(kāi)發(fā)，有時(shí)候需要返回一個(gè)流式響應(yīng)（Streaming Response） ，比如：

• 實(shí)時(shí)輸出日誌
• 逐步生成大文件（如CSV）
• 模擬實(shí)時(shí)數(shù)據(jù)推送（如SSE）
• AI 回答逐字輸出

Flask 提供了Response類來(lái)支持流式輸出。下面是一個(gè)簡(jiǎn)單實(shí)用的Flask 流式響應(yīng)示例，並解釋關(guān)鍵點(diǎn)。

? 基本流式響應(yīng)示例

from flask import Flask, Response

app = Flask(__name__)

def generate_stream():
    """生成器函數(shù)，逐步產(chǎn)生數(shù)據(jù)"""
    for i in range(1, 6):
        yield f"這是第{i} 條消息\n"
        # 可以加點(diǎn)延遲模擬真實(shí)場(chǎng)景# import time
        # time.sleep(1)

@app.route('/stream')
def stream():
    return Response(
        generate_stream(), # 傳入生成器mimetype='text/plain' # 指定MIME 類型)

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

? 說(shuō)明：

• generate_stream()是一個(gè)生成器函數(shù)，每次yield一條數(shù)據(jù)。
• Response(generator, mimetype=...)接收生成器作為內(nèi)容。
• mimetype='text/plain'表示返回純文本流。如果是HTML 或JSON 流，可以改成對(duì)應(yīng)類型。

? 瀏覽器中實(shí)時(shí)顯示（使用text/event-stream）

如果你想在瀏覽器中實(shí)現(xiàn)類似Server-Sent Events (SSE)的效果，可以用text/event-stream ：

 def generate_sse():
    for i in range(1, 6):
        yield f"data: 消息{i}\n\n" # 注意格式：data: 內(nèi)容\n\n
        # 模擬延遲import time
        time.sleep(1)

@app.route('/sse')
def sse():
    return Response(
        generate_sse(),
        mimetype='text/event-stream'
    )

前端接收示例（HTML JS）：

 <script>
const evtSource = new EventSource("/sse");
evtSource.onmessage = function(event) {
    console.log("收到:", event.data);
    document.body.innerHTML = `<p>${event.data}</p>`;
};
</script>

注意：SSE 要求後端返回text/event-stream ，且每條消息以data: ...\n\n結(jié)尾。

? 實(shí)際應(yīng)用場(chǎng)景：流式返回大文件

比如導(dǎo)出大量數(shù)據(jù)為CSV：

 import csv
from io import StringIO

def generate_csv():
    output = StringIO()
    writer = csv.writer(output)

    # 寫(xiě)表頭writer.writerow(['ID', 'Name', 'Email'])

    for i in range(1000):
        writer.writerow([i, f'User{i}', f'user{i}@example.com'])
        # 清空緩衝區(qū)並輸出一行output.seek(0)
        data = output.read()
        yield data
        output.seek(0)
        output.truncate(0) # 清空StringIO

@app.route('/download-csv')
def download_csv():
    headers = {
        'Content-Disposition': 'attachment; filename=users.csv',
        'Content-Type': 'text/csv'
    }
    return Response(
        generate_csv(),
        headers=headers
    )

這樣可以避免一次性加載所有數(shù)據(jù)到內(nèi)存，適合大數(shù)據(jù)量導(dǎo)出。

?? 注意事項(xiàng)

• 服務(wù)器/代理可能緩衝：Nginx、Gunicorn 默認(rèn)可能開(kāi)啟緩衝，導(dǎo)致看不到實(shí)時(shí)輸出。
  解決方法：
  • 使用gunicorn --workers=1 --worker-class=gevent或--worker-class=eventlet
  • 或設(shè)置Nginx的proxy_buffering off;
• 瀏覽器緩衝：某些瀏覽器會(huì)緩存部分流內(nèi)容才顯示，不是完全實(shí)時(shí)。
• 生成器不能拋異常：否則流中斷，最好加try/catch。

? 總結(jié)：關(guān)鍵點(diǎn)

• 用Response(generator, mimetype=...)創(chuàng)建流式響應(yīng)
• 用生成器yield分塊輸出
• text/plain用於文本流， text/event-stream用於SSE
• 大文件導(dǎo)出、AI 回復(fù)、日誌推送等場(chǎng)景很實(shí)用
• 注意部署環(huán)境的緩衝問(wèn)題

基本上就這些。 Flask 流式響應(yīng)不復(fù)雜，但能解決很多“實(shí)時(shí)輸出”的需求。

以上是Python燒瓶流響應(yīng)示例的詳細(xì)內(nèi)容。更多資訊請(qǐng)關(guān)注PHP中文網(wǎng)其他相關(guān)文章！