FastAPI+React实现JSON响应渐进式加载方案咨询

针对你的两层后端架构（认证转发后端 + 生成结果的微服务），需要逐个返回JSON对象让前端渐进渲染，以下是两种思路的落地实现及补充方案：

一、Server-Sent-Events (SSE) 方案 #

微服务端实现（生成SSE流） #

用sse-starlette直接生成SSE事件流，每个事件携带单个JSON对象：

from fastapi import FastAPI
from sse_starlette.sse import EventSourceResponse
import asyncio
import json

app = FastAPI()

async def generate_results():
    # 模拟耗时的结果生成逻辑
    for i in range(5):
        await asyncio.sleep(1)
        result = {"id": i, "data": f"result_{i}"}
        # 自定义event类型，方便前端区分事件
        yield {"event": "result", "data": json.dumps(result)}

@app.get("/stream-sse")
async def stream_sse():
    return EventSourceResponse(generate_results())

转发后端实现（代理SSE流） #

用aiohttp作为SSE客户端请求微服务，转发事件的同时处理认证：

from fastapi import FastAPI, Request
from sse_starlette.sse import EventSourceResponse
import aiohttp

app = FastAPI()

async def proxy_sse(request: Request):
    # 提取前端请求中的认证凭证（如Authorization头）
    auth_header = request.headers.get("Authorization")
    async with aiohttp.ClientSession() as session:
        async with session.get(
            "http://微服务地址/stream-sse",
            headers={"Authorization": auth_header} if auth_header else {}
        ) as resp:
            # 逐行读取微服务的SSE事件并直接转发
            async for line in resp.content:
                if line.strip():
                    yield line.decode("utf-8")

@app.get("/proxy-sse")
async def proxy_stream(request: Request):
    return EventSourceResponse(proxy_sse(request))

React前端处理 #

用浏览器原生EventSource监听事件，拿到结果立即更新渲染：

import { useEffect, useState } from 'react';

function ResultList() {
    const [results, setResults] = useState([]);

    useEffect(() => {
        const eventSource = new EventSource('/proxy-sse');
        eventSource.addEventListener('result', (event) => {
            const result = JSON.parse(event.data);
            setResults(prev => [...prev, result]);
        });

        return () => eventSource.close();
    }, []);

    return (
        <ul>
            {results.map(item => (
                <li key={item.id}>{item.data}</li>
            ))}
        </ul>
    );
}

二、NDJSON格式的StreamingResponse方案 #

微服务端实现（返回换行分隔的JSON流） #

用换行符分隔单个JSON对象（NDJSON格式），避免复杂字节流解析：

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio
import json

app = FastAPI()

async def generate_ndjson():
    for i in range(5):
        await asyncio.sleep(1)
        result = {"id": i, "data": f"result_{i}"}
        # 每个JSON对象后加换行符作为分隔
        yield f"{json.dumps(result)}\n"

@app.get("/stream-ndjson")
async def stream_ndjson():
    return StreamingResponse(generate_ndjson(), media_type="application/x-ndjson")

转发后端实现（代理NDJSON流） #

直接逐行读取微服务响应并转发，无需额外解析：

from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
import aiohttp

app = FastAPI()

async def proxy_ndjson(request: Request):
    auth_header = request.headers.get("Authorization")
    async with aiohttp.ClientSession() as session:
        async with session.get(
            "http://微服务地址/stream-ndjson",
            headers={"Authorization": auth_header} if auth_header else {}
        ) as resp:
            async for line in resp.content:
                yield line

@app.get("/proxy-ndjson")
async def proxy_stream(request: Request):
    return StreamingResponse(proxy_ndjson(request), media_type="application/x-ndjson")

React前端处理 #

通过Fetch API的ReadableStream逐块读取，分割换行符解析单个JSON：

import { useEffect, useState } from 'react';

function ResultList() {
    const [results, setResults] = useState([]);

    useEffect(() => {
        async function fetchStream() {
            const response = await fetch('/proxy-ndjson');
            const reader = response.body.getReader();
            const decoder = new TextDecoder('utf-8');
            let buffer = '';

            while (true) {
                const { done, value } = await reader.read();
                if (done) break;
                
                buffer += decoder.decode(value);
                // 按换行分割，处理完整的JSON行
                const lines = buffer.split('\n');
                buffer = lines.pop(); // 保留未完成的最后一行
                
                lines.forEach(line => {
                    if (line) {
                        const result = JSON.parse(line);
                        setResults(prev => [...prev, result]);
                    }
                });
            }
        }

        fetchStream();
    }, []);

    return (
        <ul>
            {results.map(item => (
                <li key={item.id}>{item.data}</li>
            ))}
        </ul>
    );
}

方案对比与选择 #

• SSE方案：自带重连机制，协议语义清晰，前端实现更简单，适合单向推送场景。
• NDJSON StreamingResponse：兼容性更广（无需考虑SSE的浏览器支持限制），可用常规HTTP请求处理，适合需要灵活控制流的场景。

若前端目标浏览器均为现代浏览器，优先选SSE；若需兼容更多环境或偏好常规Fetch API，选NDJSON方案。

内容的提问来源于stack exchange，提问作者Shiladitya Bose