Django使用StreamingHttpResponse实现流式响应的实现示例

from django.http import streaminghttpresponse
from django.views import view
import time

# 定义生成器函数，用于逐块生成数据
def stream_generator():
    """生成器函数，逐行返回数据（模拟实时生成内容）"""
    for i in range(5):
        # 模拟耗时操作（如读取大文件、实时计算）
        time.sleep(1)
        # 每块数据末尾加换行符，让客户端能实时看到
        yield f"这是第 {i+1} 块数据\n"

class streamview(view):
    def get(self, request):
        # 创建 streaminghttpresponse，传入生成器
        response = streaminghttpresponse(stream_generator())
        # 设置响应头，确保客户端按流处理（可选但推荐）
        response['content-type'] = 'text/plain; charset=utf-8'
        # 禁用缓存，确保数据实时推送
        response['cache-control'] = 'no-cache'
        # 添加流式传输 headers，禁用缓冲
        # 注意：transfer-encoding 和 connection 是 hop-by-hop headers，由 wsgi 服务器自动处理，不能在应用层设置
        # response['cache-control'] = 'no-cache, no-store, must-revalidate'
        # response['pragma'] = 'no-cache'
        # response['expires'] = '0'
        # response['x-accel-buffering'] = 'no'  # 禁用 nginx 缓冲

        # logger.info(f"[streamingview] streaminghttpresponse 已创建，headers 已配置用于流式传输")
        return response

from django.http import streaminghttpresponse
from django.views import view
import os

def file_stream_generator(file_path, chunk_size=8192):
    """读取文件并逐块生成数据"""
    with open(file_path, 'rb') as f:
        while chunk := f.read(chunk_size):
            yield chunk

class largefiledownloadview(view):
    def get(self, request):
        # 替换为你的大文件路径
        file_path = "/path/to/your/large_file.zip"
        if not os.path.exists(file_path):
            return httpresponse("文件不存在", status=404)
        
        # 获取文件名
        file_name = os.path.basename(file_path)
        # 创建流式响应
        response = streaminghttpresponse(
            file_stream_generator(file_path),
            content_type='application/octet-stream'  # 二进制文件类型
        )
        # 设置下载头，指定文件名
        response['content-disposition'] = f'attachment; filename="{file_name}"'
        # 设置分块传输编码
        response['transfer-encoding'] = 'chunked'
        return response

from django.http import streaminghttpresponse
from django.views import view
import time
import json

def sse_generator():
    """生成 sse 格式的实时数据"""
    for i in range(10):
        time.sleep(1)
        # sse 格式要求：data: 内容\n\n
        data = json.dumps({"progress": i * 10, "message": f"处理中 {i*10}%"})
        yield f"data: {data}\n\n"

class sseview(view):
    def get(self, request):
        response = streaminghttpresponse(sse_generator())
        # sse 专用 content-type
        response['content-type'] = 'text/event-stream'
        # 禁用缓存和超时
        response['cache-control'] = 'no-cache'
        response['x-accel-buffering'] = 'no'  # 禁用 nginx 缓冲（关键）
        return response

from django.urls import path
from .views import streamview, largefiledownloadview, sseview

urlpatterns = [
    path('stream/', streamview.as_view(), name='stream'),
    path('download-large-file/', largefiledownloadview.as_view(), name='download'),
    path('sse/', sseview.as_view(), name='sse'),
]