NodeJS中流式读取大Firebase JSON文件并逐对象处理

核心思路 #

直接用JSON.parse加载大文件会占用大量内存，我们可以借助流式JSON解析库，仅在内存中保留当前正在处理的单个顶级对象（比如obj1、obj2），处理完成后立即释放，大幅降低内存负载。推荐使用stream-json库，它专为流式处理JSON设计，能高效提取指定层级的JSON节点。

实现步骤 #

1. 安装依赖 #

npm install stream-json fs

2. 完整代码示例 #

const { createReadStream } = require('fs');
const { parser } = require('stream-json');
const { streamValues } = require('stream-json/streamers/StreamValues');

// 单个对象的处理逻辑
async function processObject(key, data) {
  console.log(`开始处理对象: ${key}`);
  // 这里替换成你的业务逻辑，比如：
  // 解析日期格式、统计访问数据、写入数据库等
  console.log(`对象 ${key} 的访问次数: ${data.numberOfVisits}`);
  // 模拟异步操作（如写入外部存储）
  await new Promise(resolve => setTimeout(resolve, 100));
  console.log(`完成处理对象: ${key}\n`);
}

// 流式读取并处理整个JSON文件
async function processLargeJsonFile(filePath) {
  const readStream = createReadStream(filePath);
  const jsonParser = parser();
  const valueStream = streamValues(jsonParser);

  for await (const { key, value } of valueStream) {
    // 仅处理顶级键对应的对象（跳过非对象类型的顶级节点）
    if (typeof value === 'object' && value !== null) {
      await processObject(key, value);
    }
  }

  console.log('所有对象处理完成');
}

// 执行处理流程
processLargeJsonFile('./firebase-data.json')
  .catch(err => console.error('处理出错:', err));

3. 代码说明 #

• createReadStream: 从目标文件创建可读流，逐块读取数据，不会一次性将整个文件加载到内存。
• parser(): stream-json的核心解析器，将流式二进制数据转换为JSON节点流。
• streamValues(): 过滤并提取JSON中的值节点，自动处理嵌套结构，直接返回顶级键值对。
• for await...of: 异步遍历流中的每个顶级对象，确保前一个对象处理完成后再处理下一个，避免内存堆积。

备选方案：使用JSONStream #

如果更习惯JSONStream库，也可以用以下实现：

首先安装依赖：

npm install JSONStream fs

const { createReadStream } = require('fs');
const JSONStream = require('JSONStream');

async function processObject(key, data) {
  // 同上述处理逻辑
}

function processLargeJsonFile(filePath) {
  return new Promise((resolve, reject) => {
    createReadStream(filePath)
      .pipe(JSONStream.parse('$*')) // 捕获所有顶级键值对，格式为[key, value]
      .on('data', async ([key, value]) => {
        const stream = this;
        stream.pause(); // 暂停流，避免处理不及时导致内存堆积
        try {
          await processObject(key, value);
        } catch (err) {
          stream.destroy(err);
          return;
        }
        stream.resume(); // 恢复流继续读取下一个对象
      })
      .on('end', () => {
        console.log('所有对象处理完成');
        resolve();
      })
      .on('error', err => {
        console.error('处理出错:', err);
        reject(err);
      });
  });
}

注意事项 #

• 若处理函数包含IO操作（如写入数据库），务必用await等待操作完成，避免流的读取速度超过处理速度导致内存占用飙升。
• 流处理过程中需及时捕获错误并销毁流，防止程序意外崩溃。

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