Node.js流处理通过可读、可写、双工和转换流实现高效数据处理，利用pipe()方法连接流并自动管理背压，结合stream.pipeline进行错误处理，适用于大文件、网络通信等场景，提升内存和时间效率。
在Node.js中处理数据，尤其当面对大量信息时，直接把所有内容加载到内存里往往不是一个好主意，甚至可能导致程序崩溃。流（Stream）就是Node.js为此提供的一套强大机制，它允许你以小块、渐进的方式处理数据，而不是一次性全部读取或写入。这就像一条管道，数据一点点流过，一边接收一边处理，极大地提升了效率和内存管理。
使用Node.js流处理数据的核心在于理解其四种基本类型：可读流（Readable）、可写流（Writable）、双工流（Duplex）和转换流（Transform）。它们都继承自
EventEmitter
data
end
error
pipe()
一个典型的场景是读取一个大文件，对内容进行某种处理，然后写入另一个文件。
const fs = require('fs');
const zlib = require('zlib'); // 用于压缩的转换流
// 创建一个可读流，从input.txt读取数据
const readableStream = fs.createReadStream('input.txt');
// 创建一个可写流，将处理后的数据写入output.txt
const writableStream = fs.createWriteStream('output.txt');
// 创建一个转换流，这里我们用gzip来压缩数据
const transformStream = zlib.createGzip();
// 将可读流、转换流和可写流连接起来
// 数据从input.txt流出，经过gzip压缩，最后流入output.txt
readableStream
.pipe(transformStream) // 将可读流连接到转换流
.pipe(writableStream) // 将转换流连接到可写流
.on('finish', () => {
console.log('文件压缩并写入完成！');
})
.on('error', (err) => {
console.error('流处理过程中发生错误:', err);
});
// 也可以直接将一个可读流pipe到一个可写流，不经过转换
// fs.createReadStream('input.txt').pipe(fs.createWriteStream('output_copy.txt'));这个例子展示了流处理的简洁和强大，
pipe()
我记得刚开始写Node.js的时候，总觉得把文件一股脑读进来最省事，用
fs.readFile
fs.readFileSync
首先，内存效率是最大的考量。传统方式（比如
fs.readFile
其次，时间效率也有显著提升。因为流是边读边处理的，程序不需要等到整个文件读取完毕才能开始处理。这意味着用户可以更快地看到部分结果，或者数据可以更快地被转发到下一个处理环节。比如，一个大文件上传，用户不需要等到文件完全上传完毕才能开始处理，你可以一边接收一边进行病毒扫描或数据解析。
再者，背压（Backpressure）机制是流的一个智能特性。想象一下，一个数据生产者（比如一个快速读取文件的流）比一个数据消费者（比如一个缓慢写入数据库的流）快很多。如果没有背压机制，生产者会不断地把数据推给消费者，导致消费者内部的缓冲区迅速膨胀，最终耗尽内存。Node.js的流通过
pipe()
最后，流的可组合性非常强。通过
pipe()
所以，当你在Node.js中遇到文件I/O、网络通信或者任何涉及大量数据处理的场景时，第一个应该考虑的就是流。它能帮你构建出更健壮、更高效的应用。
理解Node.js流的四种核心类型是掌握流处理的关键。它们各自有明确的职责和适用场景，但通过
pipe()
可读流（Readable Stream）
fs.createReadStream()
http.IncomingMessage
const fs = require('fs');
const readable = fs.createReadStream('large_file.txt');
readable.on('data', (chunk) => {
console.log(`收到 ${chunk.length} 字节数据`);
// 处理数据块
});
readable.on('end', () => {
console.log('文件读取完毕');
});
readable.on('error', (err) => {
console.error('读取错误:', err);
});可写流（Writable Stream）
fs.createWriteStream()
http.ServerResponse
net.Socket
const fs = require('fs');
const writable = fs.createWriteStream('output.txt');
writable.write('Hello, Node.js Streams!\n');
writable.write('This is some more data.\n');
writable.end('End of data.'); // 调用end()表示所有数据已写入
writable.on('finish', () => {
console.log('数据写入完成');
});
writable.on('error', (err) => {
console.error('写入错误:', err);
});双工流（Duplex Stream）
net.Socket
stdio
process.stdin
process.stdout
process.stderr
stdio
转换流（Transform Stream）
职责： 是一种特殊的双工流，它在数据从可读端流向可写端时，会对数据进行修改或转换。它就像管道中间的一个过滤器或处理器。
应用场景：
zlib.createGzip()
zlib.createGunzip()
crypto.createCipher()
crypto.createDecipher()
示例：
const { Transform } = require('stream');
// 创建一个将所有文本转换为大写的转换流
const upperCaseTransform = new Transform({
transform(chunk, encoding, callback) {
// chunk是Buffer，需要先转换为字符串，处理后再转回Buffer
this.push(chunk.toString().toUpperCase());
callback(); // 告诉流处理完成
}
});
process.stdin // 从标准输入读取
.pipe(upperCaseTransform) // 经过大写转换
.pipe(process.stdout); // 输出到标准输出
// 尝试在终端输入一些小写字母，会看到大写输出这四种流类型构成了Node.js流处理的骨架，理解它们各自的特点和如何通过
pipe()
在Node.js流处理中，错误处理和背压管理是确保应用健壮性和稳定性的两大关键点。我曾经因为疏忽这两点，导致生产环境出现内存泄漏和程序崩溃，所以对此感触很深。
1. 有效地处理流中的错误
流操作本质上是异步的，错误随时可能发生，比如文件不存在、网络中断、数据解析失败等。如果不妥善处理，一个未捕获的错误足以让整个Node.js进程崩溃。
'error'
EventEmitter
'error'
const fs = require('fs');
const readable = fs.createReadStream('non_existent_file.txt');
readable.on('error', (err) => {
console.error('可读流发生错误:', err.message);
// 在这里进行错误恢复或优雅地关闭应用
});
// 如果不监听，当文件不存在时，程序会直接崩溃pipe()
pipe()
pipe()
'error'
const fs = require('fs');
const zlib = require('zlib');
const readable = fs.createReadStream('input.txt');
const transform = zlib.createGzip();
const writable = fs.createWriteStream('output.gz');
readable.on('error', (err) => console.error('读取流错误:', err.message));
transform.on('error', (err) => console.error('转换流错误:', err.message));
writable.on('error', (err) => console.error('写入流错误:', err.message));
readable.pipe(transform).pipe(writable);然而，这种方式有点冗余，尤其当链条很长时。Node.js v10 引入的
stream.pipeline
const { pipeline } = require('stream');
const fs = require('fs');
const zlib = require('zlib');
pipeline(
fs.createReadStream('input.txt'),
zlib.createGzip(),
fs.createWriteStream('output.gz'),
(err) => {
if (err) {
console.error('管道处理失败:', err.message);
// 在这里进行统一的错误处理和资源清理
} else {
console.log('管道处理成功！');
}
}
);强烈推荐在生产环境中使用
stream.pipeline
2. 有效地管理背压
背压这东西，我刚开始理解的时候觉得有点抽象，但一旦遇到生产环境里消费者处理不过来导致内存飙升，你就知道它有多重要了。背压是指当数据生产者比消费者快时，消费者向生产者发出的“慢一点”信号。
pipe()
pipe()
highWaterMark
lowWaterMark
'drain'
手动管理（了解原理）： 尽管
pipe()
writable.write(chunk)
false
'drain'
readable.read()
null
'readable'
'data'
// 手动处理背压的简化示例
const fs = require('fs');
const readable = fs.createReadStream('input.txt');
const writable = fs.createWriteStream('output.txt');readable.on('data', (chunk) => { const canContinue = writable.write(chunk); if (!canContinue) { console.log('写入流缓冲区已满，暂停读取...'); readable.pause(); // 暂停可读流 } });
writable.on('drain', () => { console.log('写入流缓冲区已清空，恢复读取...'); readable.resume(); // 恢复可读流 });
readable.on('end', () => { writable.end(); });
writable.on('finish', () => { console.log('文件复制完成'); });
手动处理背压的代码明显比`pipe()`复杂得多，所以除非有特殊需求，否则应优先使用`pipe()`或`stream.pipeline`。
总结来说，对于错误处理，始终监听
'error'
stream.pipeline
pipe()
有时候，Node.js内置的流和
zlib
crypto
1. 继承核心流类
编写自定义流的起点是继承Node.js的
stream
stream.Readable
stream.Writable
stream.Transform
自定义可读流 (stream.Readable
_read(size)
this.push(chunk)
this.push(null)
const { Readable } = require('stream');
class MyCustomReadable extends Readable {
constructor(options) {
super(options);
this.data = ['Hello', 'World', 'Node.js', 'Stream'];
this.index = 0;
}
_read(size) {
if (this.index < this.data.length) {
const chunk = this.data[this.index];
this.push(Buffer.from(chunk + '\n')); // 每次推一个数据块
this.index++;
} else {
this.push(null); // 没有更多数据时，推入 null 表示流结束
}
}
}
const myReadable = new MyCustomReadable();
myReadable.pipe(process.stdout); // 将自定义可读流输出到标准输出自定义可写流 (stream.Writable
_write(chunk, encoding, callback)
chunk
encoding
callback
const { Writable } = require('stream');
class MyCustomWritable extends Writable {
constructor(options) {
super(options);
this.receivedData = [];
}
_write(chunk, encoding, callback) {
console.log(`接收到数据: ${chunk.toString()}`);
this.receivedData.push(chunk.toString());
// 模拟异步操作
setTimeout(() => {
callback(); // 必须调用 callback，否则流会暂停
}, 100);
}
_final(callback) { // 当所有数据都写入且流关闭时调用
console.log('所有数据已写入，最终结果:', this.receivedData.join(''));
callback();
}
}
const myWritable = new MyCustomWritable();
process.stdin.以上就是怎样使用Node.js流处理数据？的详细内容，更多请关注php中文网其它相关文章！
每个人都需要一台速度更快、更稳定的 PC。随着时间的推移，垃圾文件、旧注册表数据和不必要的后台进程会占用资源并降低性能。幸运的是，许多工具可以让 Windows 保持平稳运行。
广告
收藏
收藏
收藏
Copyright 2014-2025 https://www.php.cn/ All Rights Reserved | php.cn | 湘ICP备2023035733号
672
