NodeJS：如何从两个文件中读取并使用管道写入单个输出文件？

Question

上下文

我正在使用事件流模块来帮助我读取和写入这些我希望返回结果文件的本地文件。长话短说，我期望的 2 个输入文件（通过 express API 作为multipart/form-data 发送）的大小可以超过 200MB，其中包含一个条目列表（每行 1 个）。我想做的是将这些条目组合成以下格式<entry1>:<entry2>，其中entry1 是第一个文件中的条目，entry2 是第二个文件中的条目。我之前以一种能够在内存中存储和返回输入/输出的方式执行此操作，但是由于我的应用程序服务器上的内存空间非常有限，我的堆内存不足。我读到我可以使用事件流和管道逐行读取每个文件并输出到文件，而不是使用读取流输出到内存中的大字符串。问题是我似乎无法以正确的方式/时间解决问题，以便准备好将生成的输出文件发送回调用者。

到目前为止我所拥有的

到目前为止，我所做的工作是我得到了我期望的正确文件输出，但是，这似乎是一个异步问题，因为我在文件实际完成写入/保存之前解决了承诺。请在下面查看我的代码...

const fs = require('fs');
const es = require('event-stream');
const uuid = require('uuid');

const buildFile = async (fileOne, fileTwo) =>
    await new Promise((resolve, reject) => {
        try {
            // Output stream
            let fileID = uuid.v4();
            let outStream = fs
                .createWriteStream(`files/outputFile-${fileID}.txt`, {
                    flags    : 'a',
                    encoding : 'utf-8'
                });

            let fileOneRS = fs
                .createReadStream(fileOne.path, {
                    flags    : 'r',
                    encoding : 'utf-8'
                })
                .pipe(es.split())
                .pipe(
                    es.mapSync((lineOne) => {
                        fileOneRS.pause();

                        let fileTwoRS = fs
                            .createReadStream(fileTwo.path, {
                                flags    : 'r',
                                encoding : 'utf-8'
                            })
                            .pipe(es.split())
                            .pipe(
                                es.mapSync((lineTwo) => {
                                    fileTwoRS.pause();

                                    // Write combo to file
                                    outStream.write(`${lineOne}:${lineTwo}\n`);

                                    fileTwoRS.resume();
                                })
                            );

                        fileOneRS.resume();
                    })
                ); // This is where I have tried doing .on('end', () => resolve), but it also does not work :(
        } catch (err) {
            reject(err);
        }
    });

注意：这个函数是从另一个服务函数调用的，如下：

buildFile(fileOne, fileTwo)
    .then((result) => {
        resolve(result);
    })
    .catch((err) => {
        console.log(err);
        reject(err);
    });

作为一个新手 Javascript 开发人员，甚至是 NodeJS 的新手，我已经坚持尝试自己解决这个问题超过 2 周了。如果有人能够提供帮助，我将不胜感激！

谢谢????

Answer 1

编辑：更新代码以符合 OP 的预期输出。

一旦写入流完成，应调用 promise 的 resolve() 函数。 OP sn-p 中提供的注释表明 resolve 函数可能在耗尽 fileOneRS 时被调用（在 pipe() 链的末尾）。

不是为第一个文件中的每一行创建一个新的读取流，代码应该只实例化一次读取流。

以下示例说明如何重构此代码流，使其每行仅读取一次，并逐行连接文件 A 和 B 中的行：

import stream from "stream";
import util from "util";
import readline from "readline";
import fs from "fs";
import os from "os";

/** Returns a readable stream as an async iterable over text lines */
function lineIteratorFromFile( fileStream ){
  return readline.createInterface({
    input: fileStream,
    crlfDelay: Infinity
  })
}

// Use stream.pipeline to handle errors and to stream the combined output
// to a Writable stream. The promise will resolve once the data has finished
// writing to the output stream.
await util
  .promisify(stream.pipeline)(
    async function*(){
      for await ( const lineA of lineIteratorFromFile(fs.createReadStream( "./in1.txt" ))){
        for await (const lineB of lineIteratorFromFile(fs.createReadStream( "./in2.txt" ))){
          yield `${lineA}: ${lineB}${os.EOL}`
        }
      }
    },
    fs.createWriteStream( outputFile )
  );

NodeJS v13+ 的可运行示例可在下面折叠的 sn-p 中找到：

// in1.txt:
foo1
foo2

// in2.txt:
bar1
bar2

// out.txt (the file created by this script, with expected output):
foo1: bar1
foo1: bar2
foo2: bar1
foo2: bar2

// main.mjs:
import stream from "stream";
import util from "util";
import readline from "readline";
import fs from "fs";
import os from "os";

/** Returns a readable stream as an async iterable over text lines */
function lineIteratorFromFile( fileStream ){
  return readline.createInterface({
input: fileStream,
crlfDelay: Infinity
  })
}

(async ()=>{
  await util
.promisify(stream.pipeline)(
  async function*(){
    for await ( const lineA of lineIteratorFromFile(fs.createReadStream( "./in1.txt" ))){
      for await (const lineB of lineIteratorFromFile(fs.createReadStream( "./in2.txt" ))){
        yield `${lineA}: ${lineB}${os.EOL}`
      }
    }
  },
  fs.createWriteStream( "./out.txt" )
);
})()
  .catch(console.error);