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

更新日期: 2022-04-13阅读: 1.6k标签: node

将多个文件合并为一个文件,常见的场景是类似于大文件分片上传,事先根据一定的文件大小拆分为多个小文件上传到服务端,最后服务端在合并起来。

怎么合并?一种简单的办法是使用 fs.readFile 读取,fs.writeFile 追加写入,这种方式是将文件数据先读入应用内存再写入,不是很推荐,Node.js 本身提供了 Stream 模块可以更好的处理这种场景。

在 Stream 中合并文件之前一个比较常用的 api 是 pipe,但是这个 API 对于错误处理不是很友好,一不小心还能搞出文件句柄内存泄漏问题。

本文先介绍 pipe 方法的使用及什么情况下会遇到文件句柄的内存泄漏问题,之后再分别介绍 Stream 合并的两种实现模式。

pipe VS pipeline

pipe

创建一个可读流 readable 和一个可写流 writeable,通过管道 pipe 将可写流绑定到可读流,一个简单的 Stream 操作就完成了。

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

readable.pipe(writeable);

pipe 方法的两个参数:

  • destination:是一个可写流对象,也就是一个数据写入的目标对象,例如,上面我们创建的 writeable 就是一个可写流对象

  • options:

    • end:读取结束时终止写入流,默认值是 true

readable.pipe(destination[, options])

默认情况下我们是不需要手动调用写入流的 end 方法关闭的。

现在我们改一下, 设置 end 为 false 写入的目标流将会一直处于打开状态, 此时就需要监听可读流的 end 事件,结束之后手动调用可写流的 end 方法结束( 为什么要这样做?下文 Stream 串行合并会再用到这一特性 )。

// readable.pipe(writeable);

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

还需要注意一点 如果可读流期间发生什么错误,则写入的目标流将不会关闭 ,例如:process.stderr 和 process.stdout 可写流在 Nodejs 进程退出前将永远不会关闭,所以 需要监听错误事件,手动关闭可写流,防止内存泄漏 。

Linux 下一切皆文件,为了测试,在创建可读流时,你可以不创建 test1.txt 文件,让可读流自动触发 error 事件并且将 writeable 的 close 方法注释掉,通过 linux 命令 ls -l /proc/${pid}/fd 查看 error 和非 error 前后的文件句柄变化。

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

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

以下为触发 error 错误下 test2.txt 这个文件 fd 将会一直打开,除非进程退出,所以重要的事情再说一遍, 如果使用 pipe 一定要做好错误监听手动关闭每个写入流 ,以防止 “ 内存泄漏 ”。

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

注意,Mac 下没有 /proc 文件,可通过 docker 测试。不想开两个终端的,也可以在程序 setInterval 定时器函数里使用 child_process 模块的 exec 函数执行 ls -l /proc/${process.pid}/fd 命令。

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

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

pipeline

Stream 模块的一个新 API pipeline 方法,添加于 Node.js v10.0,Promise 风格需要 Node.js  v15.0+ 支持。相比较于 pipe 方法增加了错误处理机制,当管道中的某个流发生错误,它会自动处理并释放掉相应的资源。

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

串行模式 Stream 合并

使用 pipe 方法实现串行模式的流合并,根据前面讲的,设置可读流的 end 为 false 保持写入流一直处于打开状态,直到所有的可读流结束(待合并的文件完成后),我们再将可写流给关闭。

  • streamMerge 函数为入口函数

  • streamMergeRecursive 函数递归调用合并文件

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

并发模式 Stream 合并

流合并也是可以采用并发模式的,核心是通过可写流的 start、end 属性控制。

start 有点类似于数据库查询的 skip,在 计算时要求文件分块的下标必须是 0、1、2... 这样的规则 ,这种方式可以不用关注每一个流分块在文件中的存储顺序,也可以将可读流传输至可写流的指定位置。

例如,有一个大文件 dec47b76e3220432100a1155eff7f402(文件 md5 后的 hash 值) 根据 chunkSize(1048576)拆分为 3 个小文件。

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

并发模式的 Stream 合并代码实现如下:

/**
* 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,
})
);
}))
}

总结

使用 pipe 时错误处理是件需要注意的事情,特别是出现这种情况 readable.pipe(a).pipe(b).pipe(writable) 其中任何一个流关闭或出错都会导致整个管道停止工作,这个时候就要销毁所有的流,这种复杂的处理起来极其麻烦, 推荐使用 stream API pipeline 处理,或使用社区 npm 库 pump。

将多个文件合并为一个文件,使用流的方式有两种:

  • 第一种是串行模式依次读取每个文件的内容,通过 pipe 方法写入可写流,直到最后一个文件读取完成关闭写入流。

  • 另一种是并发模式,核心实现是利用写入流的 start、end 属性将可读流传输至可写流的指定位置,上面的实现还可以在优化,比如控制下并发的数量。

以上文章来源于Nodejs技术栈,作者五月君

链接: https://www.fly63.com/article/detial/11343

关于 Node.js 里 ES6 Modules 的一次更新说明

关于 Node.js 里 ES6 Modules 的一次更新说明,总结来说:CommonJS 与 ES6 Modules 之间的关键不同在于代码什么时候知道一个模块的结构和使用它。

用node.js开发一个可交互的命令行应用

在这个教程中,我们会开发一个命令行应用,它可以接收一个 CSV 格式的用户信息文件,教程的内容大纲:“Hello,World”,处理命令行参数,运行时的用户输入,异步网络会话,美化控制台的输出,封装成 shell 命令,JavaScript 之外

Node启动https服务器

首先你需要生成https证书,可以去付费的网站购买或者找一些免费的网站,可能会是key或者crt或者pem结尾的。不同格式之间可以通过OpenSSL转换

nodejs 异步转同步

nodej项目在微信环境开发,nodejs的异步特效,会导致请求没有完成就执行下面的代码,出现错误。经过多方查找,可以使用async模块来异步转同步,只有前一个function执行callback,下一个才会执行。

基于node服务器的大文件(G级)上传

3G的大文件分1500个2M二进度文件,通post方法发送给node服务,服务器全部接收到文件后,进组装生成你上文件。

为什么要把 JavaScript 放到服务器端上运行?

JavaScript比C的开发门槛要低,尽管服务器端JavaScript存在已经很多年了,但是后端部分一直没有市场,JavaScript在浏览器中有广泛的事件驱动方面的应用,考虑到高性能、符合事件驱动、没有历史包袱这3个主要原因,JavaScript成为了Node的实现语言。

了解node.js事件循环

node.js的第一个基本论点是I / O的性能消耗是很昂贵。因此,使用当前编程技术的最大浪费来自于等待I / O完成。有几种方法可以处理性能影响

Node.js 应用:Koa2 使用 JWT 进行鉴权

在前后端分离的开发中,通过 Restful API 进行数据交互时,如果没有对 API 进行保护,那么别人就可以很容易地获取并调用这些 API 进行操作。那么服务器端要如何进行鉴权呢?

Node.js 前端开发指南

我们经常跟Node.js打交道,即使你是一名前端开发人员 -- npm脚本,webpack配置,gulp任务,程序打包 或 运行测试等。即使你真的不需要深入理解这些任务,但有时候你会感到困惑,会因为缺少Node.js的一些核心概念而以非常奇怪的方式来编码。

happypack提升项目构建速度

运行在 Node.js 之上的 Webpack 是单线程模型的,也就是说 Webpack 需要处理的任务需要一件件挨着做,不能多个事情一起做。happypack把任务分解给多个子进程去并发的执行,子进程处理完后再把结果发送给主进程。

点击更多...

内容以共享、参考、研究为目的,不存在任何商业目的。其版权属原作者所有,如有侵权或违规,请与小编联系!情况属实本人将予以删除!