NDJSON 流式异步迭代消费

示例说明

NDJSON 每行一个独立 JSON 记录，是日志导出与事件管道的通用格式。网络分块与行边界毫无对齐关系，一个 JSON 对象可能劈在两个 chunk 里，也可能一 chunk 含多行——缓冲区拼接再按换行切分是唯一正确姿势。流式处理让百 MB 数据的内存占用恒定在缓冲区大小级别。

技术要点

• decode 的 stream 选项正确处理跨块 UTF-8
• 缓冲区按换行切分重组逻辑行
• 单行解析失败被隔离不中断流

代码解读

ndjsonLines 是异步生成器：每个 chunk 解码后并入缓冲区，内层 while 把所有完整行 yield 出去，残余留给下一轮；流结束时做最后一次 flush 防止末行无换行符丢失。ingest 循环里 try 包住单行解析，坏行计入 failed 继续消费。

使用建议

背压由 for await 机制天然提供，消费慢时读取自动暂停。若上游是 gzip 响应需先解压再分行。长期任务要把 processed 与 failed 周期性落盘，支持断点重放。

代码示例

async function* ndjsonLines(body) {
  let buffer = '';
  const decoder = new TextDecoder();

  for await (const chunk of body) {
    buffer += decoder.decode(chunk, { stream: true });
    let newline;
    while ((newline = buffer.indexOf('\n')) !== -1) {
      const line = buffer.slice(0, newline).trim();
      buffer = buffer.slice(newline + 1);
      if (line) yield line;
    }
  }
  buffer += decoder.decode();
  if (buffer.trim()) yield buffer.trim();
}

async function ingest(url) {
  let processed = 0;
  let failed = 0;

  const res = await fetch(url, { headers: { Accept: 'application/x-ndjson' } });
  if (!res.ok) throw new Error('stream failed: ' + res.status);

  for await (const line of ndjsonLines(res.body)) {
    try {
      const record = JSON.parse(line);
      processed += 1;
      if (processed % 1000 === 0) console.log('processed', processed);
    } catch {
      failed += 1; // 单行损坏不影响整条流
    }
  }
  return { processed, failed };
}

const stats = await ingest('https://api.example.com/events/stream');
console.log('ingest done:', stats);