Node.js 流处理：高效处理大数据的艺术
什么是流？
在 Node.js 中，流（Stream）是处理大量数据的抽象接口。它允许我们逐块读取或写入数据，而不需要一次性将全部数据加载到内存中。
为什么需要流？
想象一下处理一个 10GB 的日志文件：
- 如果使用 fs.readFile，会将整个文件加载到内存中，可能导致内存溢出
- 使用流，可以逐块读取，每处理完一块就释放内存
流的四种类型
1. Readable（可读流）
用于读取数据，例如从文件或网络读取。
const fs = require('fs');
const readable = fs.createReadStream('large-file.txt');
readable.on('data', (chunk) => {
console.log(`Received ${chunk.length} bytes`);
});
readable.on('end', () => {
console.log('Finished reading');
});
2. Writable（可写流）
用于写入数据，例如写入文件或发送到网络。
const fs = require('fs');
const writable = fs.createWriteStream('output.txt');
writable.write('Hello, ');
writable.write('World!');
writable.end();
3. Duplex（双工流）
既可以读取也可以写入，例如 TCP socket。
const net = require('net');
const server = net.createServer((socket) => {
socket.write('Hello from server');
socket.on('data', (data) => {
console.log(`Received: ${data}`);
});
});
4. Transform（转换流）
在读取和写入之间进行数据转换，例如压缩、加密。
const { Transform } = require('stream');
const upperCase = new Transform({
transform(chunk, encoding, callback) {
this.push(chunk.toString().toUpperCase());
callback();
}
});
流的核心概念
背压（Backpressure）
当写入速度慢于读取速度时，数据会在内存中堆积，导致内存溢出。流自动处理背压问题。
readable.on('data', (chunk) => {
if (!writable.write(chunk)) {
readable.pause();
}
});
writable.on('drain', () => {
readable.resume();
});
Pipe（管道）
使用 pipe 方法可以自动处理背压，是推荐的数据传输方式。
const fs = require('fs');
const zlib = require('zlib');
fs.createReadStream('input.txt')
.pipe(zlib.createGzip())
.pipe(fs.createWriteStream('input.txt.gz'));
实战：创建自定义流
创建自定义可读流
const { Readable } = require('stream');
class NumberStream extends Readable {
constructor(max) {
super({ objectMode: true });
this.max = max;
this.current = 1;
}
_read() {
if (this.current <= this.max) {
this.push(this.current++);
} else {
this.push(null);
}
}
}
const stream = new NumberStream(5);
stream.on('data', (num) => console.log(num));
创建自定义转换流
const { Transform } = require('stream');
class JSONParser extends Transform {
constructor() {
super({ readableObjectMode: true });
this.buffer = '';
}
_transform(chunk, encoding, callback) {
this.buffer += chunk;
let index;
while ((index = this.buffer.indexOf('\\n')) !== -1) {
const line = this.buffer.slice(0, index);
this.buffer = this.buffer.slice(index + 1);
try {
this.push(JSON.parse(line));
} catch (e) {
console.error('Invalid JSON:', line);
}
}
callback();
}
_flush(callback) {
if (this.buffer) {
try {
this.push(JSON.parse(this.buffer));
} catch (e) {
console.error('Invalid JSON:', this.buffer);
}
}
callback();
}
}
流的高级用法
并发流处理
const { pipeline, Transform } = require('stream');
const fs = require('fs');
const processor = new Transform({
transform(chunk, encoding, callback) {
const result = processChunk(chunk);
callback(null, result);
}
});
pipeline(
fs.createReadStream('input.txt'),
processor,
fs.createWriteStream('output.txt'),
(err) => {
if (err) {
console.error('Pipeline failed:', err);
} else {
console.log('Pipeline succeeded');
}
}
);
流与 Promise 结合
const { pipeline } = require('stream/promises');
const fs = require('fs');
async function processFile() {
try {
await pipeline(
fs.createReadStream('input.txt'),
fs.createWriteStream('output.txt')
);
console.log('Processing complete');
} catch (err) {
console.error('Error:', err);
}
}
流在实际项目中的应用
场景一：日志处理
const fs = require('fs');
const { createInterface } = require('readline');
const rl = createInterface({
input: fs.createReadStream('access.log'),
crlfDelay: Infinity
});
rl.on('line', (line) => {
const log = parseLog(line);
if (log.statusCode >= 400) {
console.log('Error:', line);
}
});
场景二：数据转换
const csv = require('csv-parser');
const fs = require('fs');
fs.createReadStream('data.csv')
.pipe(csv())
.on('data', (row) => {
const json = transformRow(row);
writeToDatabase(json);
})
.on('end', () => {
console.log('CSV parsing complete');
});
场景三：HTTP 响应流
const http = require('http');
const fs = require('fs');
http.createServer((req, res) => {
const stream = fs.createReadStream('large-file.zip');
res.writeHead(200, { 'Content-Type': 'application/zip' });
stream.pipe(res);
}).listen(3000);
性能优化建议
1. 使用适当的 highWaterMark
const stream = fs.createReadStream('file.txt', {
highWaterMark: 64 * 1024 // 64KB
});
2. 避免不必要的数据转换
尽可能在流中直接处理数据，避免多次转换。
3. 使用对象模式
对于非二进制数据，使用 objectMode: true 可以提高可读性。
总结
Node.js 流是处理大数据的利器，掌握流的使用能够：
从日志分析到文件处理，从数据转换到 HTTP 响应，流的应用无处不在。深入理解流的原理和用法，将使你成为更优秀的 Node.js 开发者。
