Node.js中await语句未按预期工作的问题排查

问题根源 #

readline的line事件是非阻塞触发的——哪怕你在回调里写了await，它也不会停下来等待异步操作完成，会继续读取文件内容、触发新的line事件。结果就是你还没处理完当前20行的chunk，lines数组就已经被塞满了整个文件的内容；等文件读完触发close事件时，又会把所有行再处理一遍，导致重复执行。

方案1：顺序处理每个chunk #

核心思路是处理chunk时暂停读取流，处理完成后再恢复，确保上一批处理完才会读取下一批：

const fs = require('fs');
const readline = require('readline');
const filePath = 'the/path/to/the/file';
const linesPerChunk = 20;
const readStream = fs.createReadStream(filePath, { encoding: 'utf8' });
const rl = readline.createInterface({
  input: readStream,
  crlfDelay: Infinity
});

let lines = [];

rl.on('line', async (line) => {
  // 如果当前流处于暂停状态，先暂存行（避免漏读）
  if (rl.isPaused()) {
    lines.push(JSON.parse(line));
    return;
  }

  lines.push(JSON.parse(line));

  if (lines.length === linesPerChunk) {
    rl.pause(); // 暂停读取，阻止新的line事件触发
    await processLines([...lines]); // 传数组副本，避免后续修改干扰
    lines = [];
    rl.resume(); // 恢复读取下一批
  }
});

rl.on('close', async () => {
  if (lines.length > 0) {
    await processLines(lines);
  }
});

async function processLines(lines) {
  console.log("process lines");
  try {
    const returnData = await someaction();
    console.log(returnData);
    console.log('Done for a chunk.');
    return returnData;
  } catch (error) {
    console.error(error);
    // 出错也要恢复流，避免程序卡死
    rl.resume();
  }
}

方案2：并行处理chunk（无重复） #

如果想提高效率、同时处理多个chunk，用for await...of遍历readline可以更优雅地控制节奏，保证每行只被加入一个chunk：

const fs = require('fs');
const readline = require('readline');
const filePath = 'the/path/to/the/file';
const linesPerChunk = 20;

async function processFile() {
  const readStream = fs.createReadStream(filePath, { encoding: 'utf8' });
  const rl = readline.createInterface({
    input: readStream,
    crlfDelay: Infinity
  });

  let lines = [];
  const tasks = [];

  // 用for await...of遍历，天然支持异步控制
  for await (const line of rl) {
    lines.push(JSON.parse(line));
    if (lines.length === linesPerChunk) {
      // 启动并行任务，不等待，继续收集下一批
      tasks.push(processLines([...lines]));
      lines = [];
    }
  }

  // 处理剩余的不足一批的行
  if (lines.length > 0) {
    tasks.push(processLines(lines));
  }

  // 等待所有并行任务完成
  await Promise.all(tasks);
}

async function processLines(lines) {
  console.log("process lines");
  try {
    const returnData = await someaction();
    console.log(returnData);
    console.log('Done for a chunk.');
    return returnData;
  } catch (error) {
    console.error(error);
  }
}

// 启动文件处理
processFile().catch(console.error);

方案对比 #

• 顺序处理：严格按文件顺序执行，适合对处理顺序有要求的场景（比如依赖上一批处理结果的操作）
• 并行处理：同时运行多个chunk的处理任务，效率更高，但要确保someaction是线程安全的，且不依赖处理顺序

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