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'),
]