nodejs createWriteStream 嵌套在 for答案

【问题标题】：nodejs createWriteStream in nested fornodejs createWriteStream 嵌套在 for
【发布时间】：2022-03-08 23:21:12
【问题描述】：

我需要生成大量字符串值组合，几十亿行，并将它们全部写入一个大文件到磁盘中，而不会阻塞进程。

此时它一直有效，直到文件写入缓冲区被填满，对于少量单词列表，当我尝试使用大量记录时，输出文件仍然为空。


const fs = require('fs');

const words = fs.readFileSync('./1024words.txt').toString().split("\n");
//create array of 1024 words

let fileout
    , row = ''
    , wordA, wordB, wordC
    , totRows = Math.pow(words.length,3)*3  //combination of 3 words for 3 columns of output
    , count = 0;

for (let indexA = 0; indexA < words.length; indexA++) {
    wordA = words[ indexA ];

    fileout = fs.createWriteStream(`${wordA}.csv`); //new output file

    for (let indexB = 0; indexB < words.length; indexB++) {
        wordB = words[ indexB ];

        for (let indexC = 0; indexC < words.length; indexC++) {
            wordC = words[ indexC ];

            row = `${wordA} ${wordB} ${wordC}\n`;

            count++;

            process.stdout.write(`${count} of ${totRows} written: ${row}`);

            fileout.write(row);
        }
    }

    fileout.end();
    fileout.close();
}

【问题讨论】：

标签： javascript node.js nodejs-stream createwritestream

【解决方案1】：

您缺少背压处理

在此处进一步阅读

https://nodejs.org/en/docs/guides/backpressuring-in-streams/

添加此功能

 function writeToStream(stream, data) {
        const canContinue = stream.write(data);
        if (!canContinue) {
            stream.pause()
            stream.once('drain', () => {
                stream.resume()
            })
        }
    }

完整代码：

const fs = require('fs');

const words = fs.readFileSync('./1024words.txt').toString().split("\n");
//create array of 1024 words

let fileout
    , row = ''
    , wordA, wordB, wordC
    , totRows = Math.pow(words.length, 3) * 3  //combination of 3 words for 3 columns of output
    , count = 0;

for (let indexA = 0; indexA < words.length; indexA++) {
    wordA = words[indexA];

    fileout = fs.createWriteStream(`${wordA}.csv`); //new output file

    for (let indexB = 0; indexB < words.length; indexB++) {
        wordB = words[indexB];

        for (let indexC = 0; indexC < words.length; indexC++) {
            wordC = words[indexC];

            row = `${wordA} ${wordB} ${wordC}\n`;

            count++;
            writeToStream(process.stdout, `${count} of ${totRows} written: ${row}`)
            writeToStream(fileout, row)
        }
    }

    fileout.end();
    fileout.close();
}

function writeToStream(stream, data) {
    const canContinue = stream.write(data);
    if (!canContinue) {
        stream.pause()
        stream.once('drain', () => {
            stream.resume()
        })
    }
}

【讨论】：