用 Next.js 实现流式接口，React 中逐步显示返回内容

在 AI 聊天对话页面中，如果接口等整段回答生成完才返回，用户就只能一直看着 loading。使用流式响应，服务端可以把已经生成的内容发给浏览器，前端收到一部分就显示一部分，让用户看到 AI 的回答是一段一段显示出来的这种效果。

这里记录一下接口和前端的实现。用定时器模拟生成回答，观察内容逐步显示、取消请求这些过程。

Next.js Demo的优势

Next.js 的 App Router 可以通过 Route Handler 定义接口。页面和接口在同一个项目中，浏览器请求 /api/stream 就可以，不用另外启动后端服务或处理跨域。

接口返回原生的 Response，响应体可以是 ReadableStream。选择 Next.js，主要是方便把 React 页面和接口放在一起。换成 Node HTTP 服务，同样可以实现流式响应。

接口返回格式

这里使用 NDJSON，每行一个 JSON 对象：

{"text":"你好"}
{"text":"这是模拟输出"}
{"text":"输出结束"}

行尾的换行符用来区分两条记录。网络传输时，两行可能放在同一个分块里，一行也可能被拆成多个分块。

如果接口使用 SSE，需要按照 SSE 的事件格式解析，不能直接套这里的逐行 JSON 解析。

Next.js 接口实现

在 App Router 项目中创建 app/api/stream/route.js：

export const runtime = 'nodejs';
export const dynamic = 'force-dynamic';

export async function GET() {
  const encoder = new TextEncoder();
  const records = ['你好', '这是模拟输出', '输出结束'];
  let timer;

  const stream = new ReadableStream({
    start(controller) {
      let index = 0;
      timer = setInterval(() => {
        if (index === records.length) {
          clearInterval(timer);
          controller.close();
          return;
        }
        const line = JSON.stringify({ text: records[index++] }) + '\n';
        controller.enqueue(encoder.encode(line));
      }, 120);
    },
    cancel() {
      clearInterval(timer);
    },
  });

  return new Response(stream, {
    headers: {
      'Content-Type': 'application/x-ndjson; charset=utf-8',
      'Cache-Control': 'no-store',
    },
  });
}

enqueue 把一段数据加入响应流，close 表示输出结束。定时器在正常结束和取消时都要清理，否则请求结束后仍然会继续执行。

真实模型接口还需要把取消传递给上游。浏览器不再读取，不代表模型服务一定停止生成。另外，这段代码只模拟少量数据，没有处理大量输出时的背压问题。

Route Handler 定义可以参考 Next.js 文档。

前端读取响应

浏览器读取到的是字节，需要解码成文本，再按换行符拆成 JSON。创建 app/read-ndjson.mjs：

export async function readNDJSON(body, onRecord) {
  if (!body) throw new Error('没有响应体');
  const reader = body.getReader();
  const decoder = new TextDecoder();
  let pending = '';

  try {
    while (true) {
      const { done, value } = await reader.read();
      pending += done
        ? decoder.decode()
        : decoder.decode(value, { stream: true });

      let end;
      while ((end = pending.indexOf('\n')) !== -1) {
        const line = pending.slice(0, end);
        pending = pending.slice(end + 1);
        if (line.trim()) onRecord(JSON.parse(line));
      }
      if (pending.length > 1024 * 1024) {
        throw new Error('单条记录过大');
      }
      if (done) break;
    }
    if (pending.trim()) onRecord(JSON.parse(pending));
  } catch (error) {
    await reader.cancel().catch(() => {});
    throw error;
  } finally {
    reader.releaseLock();
  }
}

有两种拆分需要分别处理：

• 中文字符可能被拆开。TextDecoder 使用 stream: true，会保留不完整的字符字节，等下一个分块到达再继续解码。
• JSON 记录可能被拆开。pending 保存没有换行符的部分，等下一次读取后再拼接。

结束时调用一次 decoder.decode()，处理解码器剩余内容。如果最后一条记录没有换行，也要解析。这里另外限制了没有读完的记录长度，避免错误响应一直占用内存。

React 中显示内容和取消请求

创建 app/StreamClient.jsx：

'use client';
import { useEffect, useRef, useState } from 'react';
import { readNDJSON } from './read-ndjson.mjs';

export default function StreamClient() {
  const [text, setText] = useState('');
  const [status, setStatus] = useState('未开始');
  const current = useRef(null);

  useEffect(() => () => current.current?.abort(), []);

  async function start() {
    current.current?.abort();
    const controller = new AbortController();
    current.current = controller;
    setText('');
    setStatus('接收中');

    try {
      const response = await fetch('/api/stream', {
        signal: controller.signal,
      });
      if (!response.ok) throw new Error('HTTP ' + response.status);
      await readNDJSON(response.body, record => {
        if (current.current === controller) {
          setText(value => value + record.text + '\n');
        }
      });
      if (current.current === controller) setStatus('完成');
    } catch (error) {
      if (current.current === controller) {
        setStatus(controller.signal.aborted ? '已取消' : error.message);
      }
    }
  }

  return <section>
    <button onClick={start}>开始</button>
    <button onClick={() => current.current?.abort()}>取消</button>
    <p>{status}</p>
    <pre>{text}</pre>
  </section>;
}

然后在 app/page.jsx 中使用：

import StreamClient from './StreamClient';

export default function Page() {
  return <main><h1>流式响应</h1><StreamClient /></main>;
}

运行项目后，点开始，三条内容会依次出现，最后状态变成“完成”。点取消后显示“已取消”。

这里用 ref 保存当前请求。连续点开始时，旧请求会被取消；即使旧回调晚一点执行，也不能继续修改新请求的内容和状态。因此更新状态前还要检查 current.current === controller。

只检查 response.ok 也不够。接口可能已经返回 200，但传输到一半连接断开，后续 reader.read() 仍然会失败，所以读取过程也放在 try/catch 中。

分块解析的检查

直接用完整句子作为分块，很难检查出中文解码问题。把下面代码保存到项目根目录的 check.mjs，运行 node check.mjs，每次只发送一个字节：

import assert from 'node:assert/strict';
import { readNDJSON } from './app/read-ndjson.mjs';

const bytes = new TextEncoder().encode('{"text":"你好"}\n{"text":"结束"}');
const stream = new ReadableStream({
  start(controller) {
    for (const byte of bytes) controller.enqueue(Uint8Array.of(byte));
    controller.close();
  },
});
const records = [];
await readNDJSON(stream, record => records.push(record));
assert.deepEqual(records, [{ text: '你好' }, { text: '结束' }]);
console.log('解析通过');

这个检查同时覆盖中文被拆开、JSON 被拆开和最后一行没有换行。取消、连续请求以及中途断流，还需要通过接口和页面检查。

部署时也要检查代理是否缓冲响应。本地能逐步收到内容，经过代理后却一次性出现，可能是代理把内容攒到最后才发送。流式接口解决的是逐步传输，模型生成速度和整个请求耗时仍然需要另外测量。