node.js 中的流是一种以连续方式处理数据读写的强大方法。它们使您能够有效地处理数据，特别是在处理大量信息或 i/o 操作时。本指南将介绍流的类型、如何使用它们以及实际示例，以帮助您了解流在 node.js 中的工作原理。
流是允许您以连续方式从源读取数据或将数据写入目标的对象。它们非常适合逐个处理数据，而不是一次读取或写入整个文件或缓冲区。这在处理大型数据集时特别有用，因为它可以显着减少内存使用量。
可读流：这些流允许您读取数据。示例包括 fs.createreadstream() 和 http.incomingmessage.
可写流：这些流允许您写入数据。示例包括 fs.createwritestream() 和 http.serverresponse.
双工流：这些流可以读取和写入数据。示例包括 tcp 套接字和 net.duplex。
转换流：这是一种双工流，可以在读取或写入数据时修改数据。示例包括用于压缩的 zlib.creategzip()。
您可以使用内置的 fs 模块创建可读流来读取文件，或者使用stream.readable 创建自定义可读流。
const fs = require('fs');
// create a readable stream
const readablestream = fs.createreadstream('example.txt', { encoding: 'utf8' });
// handling the 'data' event
readablestream.on('data', (chunk) => {
console.log('new chunk received:', chunk);
});
// handling the 'end' event
readablestream.on('end', () => {
console.log('no more data to read.');
});
const { readable } = require('stream');
class myreadablestream extends readable {
constructor(options) {
super(options);
this.current = 0;
}
_read(size) {
if (this.current {
console.log('received:', chunk.tostring());
});
您可以使用 fs 模块或扩展stream.writable 类来创建可写流。
const fs = require('fs');
// create a writable stream
const writablestream = fs.createwritestream('output.txt');
// write data to the stream
writablestream.write('hello, world!\n');
writablestream.write('writing to a file using streams.\n');
// end the stream
writablestream.end(() => {
console.log('finished writing to file.');
});
const { writable } = require('stream');
class mywritablestream extends writable {
_write(chunk, encoding, callback) {
console.log('writing:', chunk.tostring());
callback(); // call when done
}
}
const mywritablestream = new mywritablestream();
mywritablestream.write('hello, world!\n');
mywritablestream.write('writing to custom writable stream.\n');
mywritablestream.end();
双工流可以同时读取和写入数据。一个常见的用例是 tcp 套接字。
const { duplex } = require('stream');
class myduplexstream extends duplex {
_read(size) {
this.push('data from duplex stream\n');
this.push(null); // no more data
}
_write(chunk, encoding, callback) {
console.log('received:', chunk.tostring());
callback();
}
}
const myduplexstream = new myduplexstream();
myduplexstream.on('data', (chunk) => {
console.log('reading:', chunk.tostring());
});
// write to the duplex stream
myduplexstream.write('hello, duplex!\n');
myduplexstream.end();
转换流对于修改流经流的数据非常有用。例如，您可以使用转换流来压缩数据。
const { transform } = require('stream');
class mytransformstream extends transform {
_transform(chunk, encoding, callback) {
const upperchunk = chunk.tostring().touppercase();
this.push(upperchunk);
callback();
}
}
const mytransformstream = new mytransformstream();
mytransformstream.on('data', (chunk) => {
console.log('transformed:', chunk.tostring());
});
// pipe data through the transform stream
process.stdin.pipe(mytransformstream).pipe(process.stdout);
流的强大功能之一是将它们通过管道连接在一起的能力。管道允许您将可读流连接到可写流，这使得传输数据变得容易。
const fs = require('fs');
// create a readable stream
const readablestream = fs.createreadstream('input.txt');
// create a writable stream
const writablestream = fs.createwritestream('output.txt');
// pipe the readable stream to the writable stream
readablestream.pipe(writablestream);
writablestream.on('finish', () => {
console.log('data has been written to output.txt');
});
可读流会发出几个重要事件来帮助您管理数据流：
const fs = require('fs');
const readablestream = fs.createreadstream('example.txt');
readablestream.on('data', (chunk) => {
console.log('received chunk:', chunk.tostring());
});
readablestream.on('end', () => {
console.log('no more data to read.');
});
readablestream.on('error', (err) => {
console.error('error occurred:', err);
});
readablestream.on('close', () => {
console.log('stream closed.');
});
可写流也会发出几个事件：
const fs = require('fs');
const writablestream = fs.createwritestream('output.txt');
writablestream.on('finish', () => {
console.log('all data has been written to output.txt');
});
writablestream.on('error', (err) => {
console.error('error occurred:', err);
});
// writing data
writablestream.write('hello, world!\n');
writablestream.write('writing to a file using streams.\n');
writablestream.end(); // call end to finish the writing process
转换流从可读流和可写流继承事件，并且它们发出：
const { Transform } = require('stream');
class MyTransformStream extends Transform {
_transform(chunk, encoding, callback) {
const upperChunk = chunk.toString().toUpperCase();
this.push(upperChunk);
callback();
}
}
const myTransformStream = new MyTransformStream();
myTransformStream.on('data', (chunk) => {
console.log('Transformed chunk:', chunk.toString());
});
myTransformStream.on('end', () => {
console.log('No more data to transform.');
});
myTransformStream.on('error', (err) => {
console.error('Error occurred:', err);
});
// Write data to the transform stream
myTransformStream.write('Hello, World!\n');
myTransformStream.write('Transforming this text.\n');
myTransformStream.end(); // End the stream
node.js 中的流提供了一种强大且高效的方式来连续处理数据。它们允许您逐段读取和写入数据，这使得它们对于 i/o 操作和处理大型数据集特别有用。了解如何创建和使用不同类型的流，以及如何处理事件，将帮助您在 node.js 中构建更高效、可扩展的应用程序。
无论您是创建可读、可写、双工还是转换流，流 api 的灵活性都允许您以最适合应用程序需求的方式处理数据。