2

语境

我正在使用事件流模块来帮助我读取和写入这些我希望返回结果文件的本地文件。长话短说，multipart/form-data我期望的 2 个输入文件（通过 express API as 发送）的大小可以超过 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 周了。如果有人能够提供帮助，我将非常感谢这里的一些智慧！

谢谢

4

1 回答 1

1

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

resolve()写入流完成后，应调用promise函数。OP 片段中提供的注释表明 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+ 的可运行示例可在下面的折叠片段中找到：

// 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);

于 2020-07-14T09:00:43.480 回答