Node.js 流处理：高效处理大数据的艺术
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 开发者。