节点 - 可读流 pipe() 在 for 循环中覆盖以前的流

Question

我正在尝试使用以下代码将数据集合流式传输到多个文件：

for (var key in data) {
  // skip if collection length is 0
  if (data[key].length > 0) {
    // Use the key and jobId to open file for appending
    let filePath = folderPath + '/' + key + '_' + jobId + '.txt';

    // Using stream to append the data output to file, which should perform better when file gets big
    let rs = new Readable();
    let n = data[key].length;
    let i = 0;

    rs._read = function () {
      rs.push(data[key][i++]);

      if (i === n) {
        rs.push(null);
      }
    };

    rs.pipe(fs.createWriteStream(filePath, {flags: 'a', encoding: 'utf-8'}));

  }
}

但是，我最终让所有文件都填充了相同的数据，这是 data 对象中最后一个键的数组。似乎每个循环都覆盖了 reader 流，并且 pipe() 到可写流直到 for 循环完成才开始。这怎么可能？

Answer 1

所以您的代码可能无法正常工作的原因是因为 rs._read 方法是异步调用的，并且您的关键变量是函数范围的（因为 var 关键字）。

您创建的每个 rs 流都指向同一个变量，即键，在主循环结束时，每个回调都将具有相同的值。当您将 "var" 更改为 "let" 时，那么在每次迭代中都会创建新的关键变量，这将解决您的问题（_read 函数将拥有自己的关键变量副本而不是共享副本）。

如果你改变它让它应该工作。

Answer 2

发生这种情况是因为您在循环语句中定义的 key 不是块范围的。起初这不是问题，但是当您在 rs._read 函数内对其创建闭包时，所有后续流读取都使用最后一个已知值，即 data 数组的最后一个值。

在我们这样做的同时，我可以提出一些重构建议，使代码更简洁、更可重用：

const writeStream = (folderPath, index, jobId) => {
    const filePath = `${folderPath}/${index}_${jobId}.txt`;

    return fs.createWriteStream(filePath, {
        flags: 'a', encoding: 'utf-8'
    });
}

data.forEach((value, index) => {
    const length = value.length;

    if (length > 0) {
        const rs = new Readable();
        const n = length;

        let i = 0;

        rs._read = () => {
            rs.push(value[i++]);
            if (i === n) rs.push(null);
        }

        rs.pipe(writeStream(folderPath, index, jobId));
    }
});

节点 - 可读流 pipe() 在 for 循环中覆盖以前的流

Node - Readable stream pipe() overwrite previous streams in a for loop

pipe

stream

fs

node.js