节点嵌套的读取行和写入导致MaxListenersExceededWarning

Question

我想编写一个脚本，该脚本从source文件中获取行条目，并在search文件的每一行中搜索此字符串。如果有匹配项，我希望将搜索行复制到output文件。

[使用我当前的脚本，我只在输出文件中写了一行并且出现了MaxListenersExceededWarning错误。但是当我只有3个流时，怎么会有这么多错误？

任何帮助表示赞赏！ ：）

错误：

(node:34238) MaxListenersExceededWarning: Possible EventEmitter memory leak detected. 11 end listeners added to [Readable]. Use emitter.setMaxListeners() to increase limit
(node:34238) MaxListenersExceededWarning: Possible EventEmitter memory leak detected. 11 finish listeners added to [Readable]. Use emitter.setMaxListeners() to increase limit
(node:34238) MaxListenersExceededWarning: Possible EventEmitter memory leak detected. 11 error listeners added to [Readable]. Use emitter.setMaxListeners() to increase limit
(node:34238) MaxListenersExceededWarning: Possible EventEmitter memory leak detected. 11 close listeners added to [Readable]. Use emitter.setMaxListeners() to increase limit
(node:34238) MaxListenersExceededWarning: Possible EventEmitter memory leak detected. 11 readable listeners added to [Readable]. Use emitter.setMaxListeners() to increase limit

这是我的剧本：

const fs = require('fs')
const readline = require('readline')

const sourceEngramFile = './super-structures-engrams.txt'
const searchEngramConfig = './all-engram-overrides.ini'
const outputEngramConfig = './output-super-structures.ini'

async function writeLine(stream, line) {
  return new Promise(resolve => {
    stream.write(line, 'utf8', resolve)
  })
}

async function processLineByLine() {
  const sourceStream = fs.createReadStream(sourceEngramFile)
  const searchStream = fs.createReadStream(searchEngramConfig)
  const outputStream = fs.createWriteStream(outputEngramConfig)

  const sourceRL = readline.createInterface({
    input: sourceStream,
    crlfDelay: Infinity,
  })

  const searchRL = readline.createInterface({
    input: searchStream,
    crlfDelay: Infinity,
  })

  // Possibly the error is caused by these nested loops/awaits
  for await (const sourceLine of sourceRL) {
    for await (const searchLine of searchRL) {
      if (searchLine.includes(`EngramEntry_${sourceLine}`)) {
        await writeLine(outputStream, searchLine)
      }
    }
  }
}

processLineByLine()

源文件看起来像这样：

Wall_Wood
Wall_Tek
Wall_Greenhouse
...

搜索文件如下：

OverrideNamedEngramEntries=(EngramClassName="EngramEntry_Wall_Wood_C",EngramLevelRequirement=11,EngramPointsCost=7,EngramHidden=false,RemoveEngramPreReq=true)
OverrideNamedEngramEntries=(EngramClassName="EngramEntry_Hatchframe_Adobe_C",EngramLevelRequirement=16,EngramPointsCost=8,EngramHidden=false,RemoveEngramPreReq=true)
OverrideNamedEngramEntries=(EngramClassName="EngramEntry_Doorframe_Wood_C",EngramLevelRequirement=11,EngramPointsCost=6,EngramHidden=false,RemoveEngramPreReq=true)
...

Answer 1

maxListeners的问题是由于尝试为内部searchRL循环重复使用同一for对象而引起的。每次尝试使用它时，它都会将一组新的侦听器添加到基础的readStream上，最终超出发出警告的侦听器的数量。我能够在一个简单的测试应用中重现相同的警告。

也是这种情况，您的内部for循环尽管通过外部循环还是只能在第一次使用。之后，searchStream被完全消耗，并且不再产生任何输出。

您的循环此处：

  // Possibly the error is caused by these nested loops/awaits
  for await (const sourceLine of sourceRL) {
    for await (const searchLine of searchRL) {
      if (searchLine.includes(`EngramEntry_${sourceLine}`)) {
        await writeLine(outputStream, searchLine)
      }
    }
  }

试图一遍又一遍地迭代searchRL流。这是行不通的，因为底层的读取流在第一次迭代时就被完全消耗掉了，从那时起，迭代器就没有要读取的内容。同样的问题也会导致您的eventListener问题。

如果searchStream可以合理地容纳在内存中，那么您应该将其读入内存一次，以便可以在sourceStream的迭代中反复使用它。如果不是这样，并且您希望每次都从文件中进行迭代，那么您可能每次都必须通过外循环创建一个新的流和一个新的readLine对象，以便有一个新的流可以进行迭代。

为了演示我正在讨论的问题，这里有个小程序，除了尝试重复两次相同的readLine对象外，什么也不做。第一次迭代得到我们想要的数据。第二个不获取数据，因为底层的读取流已在第一次迭代中完全消耗掉了（例如，流中的文件指针指向文件的末尾，因此没有其他要读取的内容）。

const fs = require('fs');
const readline = require('readline');

async function run() {
    let rl = readline.createInterface({input: fs.createReadStream("file1.txt")});

    rl.on('error', err => {
        console.log(err);
    });

    console.log("start1");
    for await (const line of rl)  {
        console.log(line);
    }
    console.log("start2");
    for await (const line of rl)  {
        console.log(line);
    }
}

run().then(() => {
    console.log("done");
}).catch(err => {
    console.log(err);
});

不幸的是，指向文件的readStreams没有将其状态重置回文件开头的内置方法，所以我认为您只需要创建一个新的流和一个新的readLine对象或读取流一次进入内存，并在每次循环时从内存而不是文件中使用它。