Node.js 流式处理CSV与API限速的异步控制策略

const fs = require('fs');
const { parse } = require('csv-parse');
const { stringify } = require('csv-stringify');

// 辅助函数：生成指定范围内的随机整数
function randomIntFromInterval(min, max) {
  return Math.floor(Math.random() * (max - min + 1) + min);
}

// 辅助函数：创建Promise延迟
function timeout() {
  return new Promise((resolve) =>
    setTimeout(resolve, randomIntFromInterval(3000, 39990)) // 3秒到近40秒的随机延迟
  );
}

// 模拟API调用
async function callAPI(firstName) {
  console.log(`Calling API for: ${firstName}`);
  await timeout(); // 模拟API响应时间
  return `Info for ${firstName}`;
}

// 处理单行数据
async function processData(row, data) {
  const firstName = row[0];
  const infos = await callAPI(firstName);
  // 再次引入延迟，避免后续处理过快，但这里不是关键
  await timeout(); 
  const newRow = [firstName, infos];
  data.push(newRow);
  return data;
}

function processCsvWithFloodingRisk() {
  return new Promise((resolve, reject) => {
    const promises = [];
    const processedData = []; // 用于存储处理后的数据

    fs.createReadStream("./input.csv")
      .pipe(parse({ delimiter: ",", from_line: 1 }))
      .on("data", async (row) => {
        // 问题所在：on('data') 事件是异步触发的，每次触发都会立即执行
        // processData并将其Promise推入数组，而不会等待上一个Promise完成。
        // 即使这里有 await timeout()，它也只延迟了当前 on('data') 回调的后续代码，
        // 而不是阻止下一个 on('data') 事件的触发。
        promises.push(processData(row, processedData));
        await timeout(); // 这个延迟仅影响当前 on('data') 回调的内部流程
      })
      .on("end", async () => {
        console.log("Stream ended, waiting for all promises to resolve...");
        await Promise.all(promises); // 此时所有API调用可能已经并发发出
        stringify(processedData, (err, output) => {
          if (err) return reject(err);
          fs.writeFileSync("output.csv", output);
          console.log("Output written to output.csv");
          resolve();
        });
      })
      .on("error", function (error) {
        console.error("Stream error:", error.message);
        reject(error);
      });
  });
}

// 假设 input.csv 存在，内容类似：
// Name,Age
// Alice,30
// Bob,25
// Charlie,35
// ...
// processCsvWithFloodingRisk().catch(console.error);

const fs = require('fs');
const { parse } = require('csv-parse');
const { stringify } = require('csv-stringify');

// 辅助函数：生成指定范围内的随机整数
function randomIntFromInterval(min, max) {
  return Math.floor(Math.random() * (max - min + 1) + min);
}

// 辅助函数：创建Promise延迟
function timeout() {
  return new Promise((resolve) =>
    setTimeout(resolve, randomIntFromInterval(3000, 39990)) // 3秒到近40秒的随机延迟
  );
}

// 模拟API调用
async function callAPI(firstName) {
  console.log(`Calling API for: ${firstName} at ${new Date().toLocaleTimeString()}`);
  await timeout(); // 模拟API响应时间
  return `Info for ${firstName}`;
}

// 处理单行数据
async function processData(row) {
  const firstName = row[0];
  const infos = await callAPI(firstName);
  return [firstName, infos];
}

async function processCsvWithRateLimitControl() {
  const readStream = fs.createReadStream('./input.csv');
  const writeStream = fs.createWriteStream('output.csv');
  const parser = parse({ delimiter: ',', from_line: 1 });

  const processedRows = []; // 存储处理后的行数据

  // 将读取流管道到解析器
  readStream.pipe(parser);

  try {
    // 使用 for await...of 迭代解析器，它是一个异步可迭代对象
    for await (const row of parser) {
      console.log(`Processing row: ${row[0]}`);
      const newRow = await processData(row); // 顺序处理每一行，等待API调用完成
      processedRows.push(newRow);
      await timeout(); // 在处理下一行之前引入延迟，严格控制请求速率
    }

    // 所有行处理完毕后，将结果写入CSV
    stringify(processedRows, (err, output) => {
      if (err) {
        console.error("Error stringifying data:", err);
        return;
      }
      writeStream.write(output);
      writeStream.end();
      console.log('Output written to output.csv. OK');
    });

  } catch (error) {
    console.error("An error occurred during CSV processing:", error.message);
  } finally {
    // 确保流被关闭
    readStream.destroy();
    writeStream.end();
  }
}

// 启动处理流程
processCsvWithRateLimitControl().catch(console.error);

// 示例 input.csv 内容：
// Name,Age
// Alice,30
// Bob,25
// Charlie,35
// David,40
// Eve,28