节点嵌套读写导致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的问题是由于试图一次又一次地为内部for循环重用相同的searchRL对象造成的。每次您尝试使用它时，它都会向底层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对象，或者将流读入内存一次，并在每次循环时从内存中使用它，而不是从文件中使用它。