单个文件上的多个 ReadStream

Question

我有一个需要处理的大文件。该文件包含分为多个部分的标题和二进制数据。标题部分包含二进制数据部分的描述 - 定义每个部分的偏移量和长度。

我当时的想法是：

• 使用fs.openSync()获取文件描述符
• 创建ReadStream 并读取标题部分（以获取二进制数据部分的偏移量）。流以autoClose: false 打开
• 关闭“标题”ReadStream
• 创建多个ReadStream 来读取二进制数据段。每个流都有自己的start 和end，每个流都以autoClose: false 打开
• 处理完所有内容后关闭文件描述符

我试图测试这个概念，但在尝试打开第二个流时遇到了问题（在标题流关闭后）。错误是：

Uncaught Error Error: EBADF: bad file descriptor, read

这表明该文件不再打开以供读取（尽管所有流都是使用autoClose: false 创建的）

知道如何保持文件描述符打开直到手动关闭吗？

(async function () {
  // open the file for reading
  const fd = fs.openSync("c:\\some\\large\\file.txt", "r");

  // initial stream that will extract the header info
  const initStream = fs.createReadStream(null, {
    fd,
    autoClose: false,
  });

  // header info data
  const headerContent = await extractHeaderContent(initStream);

  // for test purpose
  // try and extract the header again
  const testSecondStream1 = fs.createReadStream(null, {
    fd,
    autoClose: false,
  });

  const testHeaderContent = await extractHeaderContent(initStream);

  fs.closeSync(fd);
})();

// stream the data until the header data is retrieved
async function extractHeaderContent(initStream) {
  return new Promise((resolve, reject) => {
    let content = "";

    initStream.on("data", (chunk) => {
      if (chunk.indexOf("EndHeader") > -1) {
        content += chunk.toString();
        let d = content.split("EndHeader")[0] + "EndHeader";

        // once the header info is extracted - close the stream
        initStream.close(() => resolve(d));
      }

      content += chunk.toString();
    });
  });
}

Answer 1

首先，这似乎是过度优化，将一个简单的问题变成了一个复杂的问题。为什么不在每次需要新流时只打开文件？完全没有并发症。由于您刚刚打开文件，因此在后续打开文件时，所有关于打开文件的内容都将缓存在操作系统中，因此这不应该是性能问题。

解决这个问题的简单方法是为每个流打开一个新的文件句柄，然后在完成后调用stream.destroy()。

其次，当您调用 initStream.close() 时，将关闭文件句柄（您可以看到 close 方法的代码 here）。 autoClose 仅影响流自行结束时发生的情况，而不影响手动调用 .close() 时发生的情况。

因此，您可以使用一种方案来暂停流并删除您的 data 事件处理程序。这至少会阻止 readStream 进行任何进一步的读取。我考虑调用.destroy()，但看起来它也会关闭文件句柄。

而且，仅供参考，您问题中的代码会创建testSecondStream1，但不使用它。它将initStream 传递给对extractHeaderContent() 的两个调用，我认为这只是一个愚蠢的行为，而不是你的意图（尽管当我试图运行你的代码时它让我很困惑）。

这是共享一个有效的文件句柄的复杂方法的实现。我不会以这种方式编写代码，因为它太复杂并且依赖于一些技巧才能正常工作：

const fs = require('fs');

(async function () {
    // open the file for reading
    const fd = fs.openSync("c:\some\large\file.txt", "r");

    // initial stream that will extract the header info
    const initStream = fs.createReadStream(null, {
        fd,
        autoClose: false,
        autoDestroy: false,
    });

    // header info data
    const headerContent = await extractHeaderContent(initStream);

    // for test purpose
    // try and extract the header again
    const testSecondStream1 = fs.createReadStream(null, {
        fd,
        autoClose: false,
        autoDestroy: false,
        start: 0,
    });

    const testHeaderContent = await extractHeaderContent(testSecondStream1);

    fs.closeSync(fd);
})().then(() => {
    console.log("finished");
}).catch(err => {
    console.log(err);
});

// stream the data until the header data is retrieved
async function extractHeaderContent(rStream) {
    return new Promise((resolve, reject) => {
        let content = "";

        function processData(chunk) {
            if (chunk.indexOf("EndHeader") > -1) {
                content += chunk.toString();
                let d = content.split("EndHeader")[0] + "EndHeader";

                // once the header info is extracted
                //    stop the stream from flowing
                //    and unhook from it
                rStream.pause();
                rStream.off("data", processData);
                resolve(d);

                // stop the stream from trying to clean up for itself
                rStream.destroyed = true;
                rStream.fd = null;
                return;
            }

            content += chunk.toString();
        }

        rStream.on("data", processData)
    })
}

这些是我所做的主要更改。

1. 将start: 0 添加到第二个流中，告诉它重新从头开始读取。流显然不会自动执行此操作，因此如果将 fd 留在与前一次读取不同的文件偏移量上，那么这就是第二个流开始的地方。

2. 将 autoDestroy: false 添加到两个流中。我们希望阻止流尝试关闭文件句柄本身的所有可能方式。

3. 修复对extractHeaderContent() 的第二次调用以传递testSecondStream1，而不是initStream。

4. .pause() 完成后的流，因此它不会继续读取。

5. 从流中删除我们所有的事件处理程序，以便它有资格进行垃圾收集。

6. 告诉流它已经被销毁了。这有点 hack，但显然流中的某些代码在您已经调用 fs.closeSync(fd) 之后尝试使用文件句柄，这在您完成所有处理后导致错误。我无法捕捉到是谁在这样做，所以告诉流它已经被摧毁似乎绕过它。