Node.js 进阶 - 多文件 Stream 合并，串行和并发两种模式实现

const fs = require('fs');
const readable = fs.createReadStream('./test1.txt');
const writeable = fs.createWriteStream('./test2.txt');

readable.pipe(writeable);

readable.pipe(destination[, options])

// readable.pipe(writeable);

readable.pipe(writeable, {
  end: false,
});
readable.on('end', function() {
  writeable.end('结束');
});

readable.on('error', function(err) {
  console.log('error', err);
  // writeable.close();
});

console.log(process.pid); // 打印进程 ID
setInterval(function(){}, 5000) // 让程序不中断，进程不退出

...
l-wx------ 1 root root 64 Apr 10 15:47 19 -> /root/study/test2.txt
...

const { exec } = require('child_process');

setInterval(function(){
  exec(`ls -l /proc/${process.pid}/fd`, (error, stdout, stderr) => {
    console.log(`stdout: \n`, stdout);
  })
}, 5000) // 让程序不中断，进程不退出

try {
  await pipeline(
    readable,
    writable
  );
  console.log('Pipeline succeeded.');
} catch (err) {
  console.log('error', err);
}

const fs = require('fs');
const path = require('path');

/**
 * Stream 合并
 * @param { String } sourceFileDirectory 源文件目录
 * @param { String } targetFile 目标文件
 */
function streamMerge(sourceFileDirectory, targetFile) {
  const scripts =  fs.readdirSync(path.resolve(__dirname, sourceFileDirectory)); // 获取源文件目录下的所有文件
  const fileWriteStream = fs.createWriteStream(path.resolve(__dirname, targetFile)); // 创建一个可写流

  // fs.readdir 读取出来的结果，根据具体的规则做下排序，防止因为顺序不对导致最终合并之后的文件无效。
  
  return streamMergeRecursive(scripts, fileWriteStream, sourceFileDirectory);
}

/**
 * Stream 合并的递归调用
 * @param { Array } scripts
 * @param { Stream } fileWriteStream
 */
function streamMergeRecursive(scripts=[], fileWriteStream, sourceFileDirectory) {
  // 递归到尾部情况判断
  if (!scripts.length) {
    return fileWriteStream.end("console.log('Stream 合并完成')"); // 最后关闭可写流，防止内存泄漏
  }

  const currentFile = path.resolve(__dirname, sourceFileDirectory, scripts.shift());
  const currentReadStream = fs.createReadStream(currentFile); // 获取当前的可读流

  currentReadStream.pipe(fileWriteStream, { end: false });
  currentReadStream.on('end', function() {
    streamMergeRecursive(scripts, fileWriteStream, sourceFileDirectory);
  });

  currentReadStream.on('error', function(error) { // 监听错误事件，关闭可写流，防止内存泄漏
    console.error(error);
    fileWriteStream.close();
  });
}

streamMerge('./files', './file.js');

/chunks
└── dec47b76e3220432100a1155eff7f402-1048576
    ├── dec47b76e3220432100a1155eff7f402-0
    ├── dec47b76e3220432100a1155eff7f402-1
    └── dec47b76e3220432100a1155eff7f402-2

/**
 * Stream concurrent merge
 * @param {String} sourceFileDirectory
 * @param {String} targetFile
 * @param {Number} chunkSize
 */
export const streamConcurrentMerge = async (sourceFileDirectory, targetFile, chunkSize) => {
  const filenames = await fs.readdir(sourceFileDirectory);
  
  await Promise.all(filenames.map(filename => {
    const index = filename.split('-').pop();
    const start = index * chunkSize;
    const end = (index + 1) * chunkSize;

    return pipeline(
      createReadStream(path.join(sourceFileDirectory, filename)),
      createWriteStream(targetFile, {
        start,
        end,
      })
    );
  }))
}