如何在JavaScript/TypeScript中实现嵌套结构的流式传输？

方案一：使用流式JSON解析库（推荐） #

对于嵌套结构的流式JSON解析，最可靠的方式是采用专门的流式JSON解析库，这类库能逐节点解析数据流，无需等待整个文档加载完成，还能精准识别嵌套路径下的元素。以@streamparser/json为例：

步骤1：安装依赖 #

npm install @streamparser/json

步骤2：封装API调用方法 #

import { parser } from '@streamparser/json';

// 定义API返回数据的类型
type ApiResponseTypes = {
  type1: { prop1: string; [key: string]: unknown };
  type2: { prop2: string; [key: string]: unknown };
  type3: { prop3: string; [key: string]: unknown };
};

/**
 * 流式调用API并处理不同类型的数据
 * @param endpoint API端点地址
 * @param onItem 单个数据项的处理回调
 * @param onComplete 数据流结束后的回调
 */
export async function fetchStreamedApi<T extends keyof ApiResponseTypes>(
  endpoint: string,
  onItem: (item: ApiResponseTypes[T], type: T) => void,
  onComplete?: () => void
): Promise<void> {
  const response = await fetch(endpoint, {
    headers: { Accept: 'application/json' },
  });

  if (!response.body) {
    throw new Error('响应体不可读取');
  }

  const reader = response.body.getReader();
  const decoder = new TextDecoder();

  // 创建JSON流式解析器，监听嵌套路径的元素
  const jsonParser = parser({
    onValue: (value, path) => {
      // 匹配顶级对象下数组的元素（路径格式为 ['type1', 0]）
      if (path.length === 2 && typeof path[1] === 'number') {
        const dataType = path[0] as T;
        onItem(value as ApiResponseTypes[T], dataType);
      }
    },
    onEnd: onComplete,
  });

  // 逐块读取并解析数据流
  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    jsonParser.write(decoder.decode(value, { stream: true }));
  }

  jsonParser.end();
}

步骤3：使用封装的方法 #

// 调用API并处理不同类型的数据
fetchStreamedApi('/api/data', (item, type) => {
  switch (type) {
    case 'type1':
      console.log('处理type1数据:', item.prop1);
      // 这里可以添加业务逻辑，比如渲染到页面
      break;
    case 'type2':
      console.log('处理type2数据:', item.prop2);
      break;
    case 'type3':
      console.log('处理type3数据:', item.prop3);
      break;
  }
}, () => {
  console.log('所有数据处理完成');
});

方案二：自行实现状态机解析（无依赖） #

如果不想引入第三方库，可以自己实现简易的状态机，跟踪当前解析的数组类型，逐块提取数组元素。需要注意处理JSON的语法边界（比如字符串中的特殊字符、不完整的片段等）：

type ApiResponseTypes = {
  type1: { prop1: string; [key: string]: unknown };
  type2: { prop2: string; [key: string]: unknown };
  type3: { prop3: string; [key: string]: unknown };
};

export async function fetchStreamedApi<T extends keyof ApiResponseTypes>(
  endpoint: string,
  onItem: (item: ApiResponseTypes[T], type: T) => void,
  onComplete?: () => void
): Promise<void> {
  const response = await fetch(endpoint, {
    headers: { Accept: 'application/json' },
  });

  if (!response.body) {
    throw new Error('响应体不可读取');
  }

  const reader = response.body.getReader();
  const decoder = new TextDecoder();
  let currentDataType: T | null = null;
  let buffer = '';
  // 匹配数组结束的正则
  const arrayEndRegex = /^\s*\],/;
  // 匹配数组元素的正则（非贪婪匹配对象）
  const itemRegex = /^\s*({.*?})\s*,?/gs;

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;

    buffer += decoder.decode(value, { stream: true });

    // 识别当前正在解析的数组类型
    if (!currentDataType) {
      const typeMatch = buffer.match(/"([^"]+)":\s*\[/);
      if (typeMatch) {
        currentDataType = typeMatch[1] as T;
        buffer = buffer.slice(typeMatch[0].length);
      }
    }

    // 检查当前数组是否结束，切换到下一个数组
    if (currentDataType && arrayEndRegex.test(buffer)) {
      buffer = buffer.replace(arrayEndRegex, '');
      currentDataType = null;
      continue;
    }

    // 提取当前数组中的元素
    if (currentDataType) {
      let match;
      while ((match = itemRegex.exec(buffer)) !== null) {
        try {
          const item = JSON.parse(match[1]) as ApiResponseTypes[T];
          onItem(item, currentDataType);
        } catch (e) {
          // 若片段不完整，回退匹配位置，等待下一块数据
          itemRegex.lastIndex -= match[0].length;
          break;
        }
      }
      // 更新缓冲区为未处理的剩余内容
      buffer = buffer.slice(itemRegex.lastIndex);
      itemRegex.lastIndex = 0;
    }
  }

  // 处理最后剩余的缓冲区内容
  if (currentDataType && buffer.trim()) {
    const finalItemMatch = buffer.match(/^\s*({.*?})\s*$/);
    if (finalItemMatch) {
      const item = JSON.parse(finalItemMatch[1]) as ApiResponseTypes[T];
      onItem(item, currentDataType);
    }
  }

  onComplete?.();
}

方案优缺点 #

• 优点：无需依赖第三方库，体积小
• 缺点：需要自行处理各种JSON语法边界情况（比如字符串中的逗号、转义字符），复杂场景下容易出现解析错误

总结 #

• 优先选择成熟的流式JSON解析库，避免自行实现状态机带来的边界case处理问题
• 封装的TypeScript库可以提供高抽象的API，让用户只需专注于不同类型数据的业务处理
• 确保服务器端的分块输出符合JSON语法，避免出现跨块的不完整JSON节点

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