Django流式返回多个JSON对象的正确实现方法

这个问题我之前也碰到过！直接yield字典会被Django直接转成字符串拼接，自然就成了无效的JSON格式。要实现分时段返回多个独立的JSON对象，核心是要让客户端能清晰区分每个JSON的边界，这里有两种靠谱的方案：

这是最简单的方式，遵循每行一个合法JSON对象的格式，客户端只需要按行读取就能解析每个独立的解决方案。

Django 视图实现 #

import json
import time
from django.http import HttpStreamingResponse

def streaming_solutions(request):
    def solution_generator():
        # 立即返回第一个解决方案
        first_solution = {"id": 1, "content": "第一个解决方案"}
        # 转成JSON字符串后加换行，作为分隔符
        yield json.dumps(first_solution) + "\n"
        
        # 模拟生成后续方案的耗时操作（比如调用模型、计算逻辑）
        time.sleep(2)
        
        second_solution = {"id": 2, "content": "第二个解决方案"}
        yield json.dumps(second_solution) + "\n"
        
        time.sleep(1)
        third_solution = {"id": 3, "content": "第三个解决方案"}
        yield json.dumps(third_solution) + "\n"
    
    # 指定content-type为jsonlines，让客户端明确格式
    response = HttpStreamingResponse(solution_generator(), content_type="application/jsonlines")
    # 禁用缓存，避免浏览器把整个流缓存起来
    response["Cache-Control"] = "no-cache"
    return response

客户端处理示例（JavaScript） #

用fetch读取流，按行分割解析每个JSON：

fetch('/your-stream-endpoint/')
  .then(response => {
    const reader = response.body.getReader();
    const decoder = new TextDecoder();
    let buffer = '';

    function processChunk() {
      return reader.read().then(({ done, value }) => {
        if (done) return;
        
        buffer += decoder.decode(value);
        // 按换行分割已完成的行
        const lines = buffer.split('\n');
        buffer = lines.pop(); // 保留未完成的最后一行（防止截断）

        lines.forEach(line => {
          if (line.trim()) {
            const solution = JSON.parse(line);
            console.log('收到新解决方案:', solution);
            // 在这里处理每个方案的业务逻辑
          }
        });

        return processChunk();
      });
    }

    return processChunk();
  });

这是专门为服务器向客户端推送实时数据设计的规范，格式更严谨，客户端处理也更简单，不需要手动分割流。

Django 视图实现 #

import json
import time
from django.http import HttpStreamingResponse

def sse_solutions(request):
    def event_generator():
        # SSE格式要求：每个事件以`data: `开头，结尾用两个换行分隔
        first_solution = {"id": 1, "content": "第一个解决方案"}
        yield f"data: {json.dumps(first_solution)}\n\n"
        
        time.sleep(2)
        second_solution = {"id": 2, "content": "第二个解决方案"}
        yield f"data: {json.dumps(second_solution)}\n\n"
        
        time.sleep(1)
        third_solution = {"id": 3, "content": "第三个解决方案"}
        yield f"data: {json.dumps(third_solution)}\n\n"
    
    response = HttpStreamingResponse(event_generator(), content_type="text/event-stream")
    response["Cache-Control"] = "no-cache"
    response["Connection"] = "keep-alive"  # 保持长连接
    return response

客户端处理示例（JavaScript） #

用原生EventSource监听推送：

const eventSource = new EventSource('/your-sse-endpoint/');

// 收到新事件时触发
eventSource.onmessage = function(event) {
  const solution = JSON.parse(event.data);
  console.log('收到新解决方案:', solution);
  // 处理业务逻辑
};

// 连接出错时关闭
eventSource.onerror = function(error) {
  console.error('SSE连接异常:', error);
  eventSource.close();
};

• 如果生成解决方案的逻辑很耗时，不要直接在视图里阻塞，建议把计算逻辑放到异步任务队列（比如Celery）中，生成器监听任务状态再yield结果，避免占用Django worker。
• 生产环境中，要确保你的WSGI/ASGI服务器支持流式响应（比如Gunicorn、Uvicorn），有些服务器默认可能会缓冲整个响应后再返回。

内容的提问来源于stack exchange，提问作者Rafael Marques