1. 流处理的编程日常
我在维护某电商平台的日志系统时，曾遇到过这样的场景：每天产生10GB的访问日志需要解析处理。最初的方案是直接用fs.readFile加载整个文件，结果内存直接爆掉。这就是流处理技术闪亮登场的时刻。
想象我们有个大型水电站，传统的处理方式就像把整条长江水全部抽到水库再发电，而流处理则是在河道上安装水轮机。后者既不需要超大蓄水库（内存），又能持续产生能量（数据处理）。
// 技术栈：Node.js v18 + 原生Stream模块
const fs = require('fs');
// 创建每秒10万行的测试日志
function generateLog() {
const writeStream = fs.createWriteStream('access.log');
let count = 0;
function write() {
let ok = true;
while(count < 1e5 && ok) {
ok = writeStream.write(`2023-07-20 10:00:00, GET /product/${count++}\n`);
}
if(count < 1e5) {
writeStream.once('drain', write);
}
}
write();
}
2. 流处理四重奏
2.1 可读流：数据源头活水
当处理3GB的视频文件时，直接加载会吃光内存。以下示例展示如何正确读取大文件：
const readStream = fs.createReadStream('bigfile.mp4', {
highWaterMark: 1024 * 1024 // 每次读取1MB
});
readStream.on('data', (chunk) => {
console.log(`收到${chunk.length}字节数据`);
});
readStream.on('end', () => {
console.log('文件传输完成');
});
2.2 转换流：数据美容院
需要将CSV转换为JSON时，转换流就像流水线上的装配机器人：
const { Transform } = require('stream');
class CSV2JSON extends Transform {
constructor() {
super({ objectMode: true });
this._headers = null;
}
_transform(chunk, encoding, callback) {
const rows = chunk.toString().split('\n');
if(!this._headers) {
this._headers = rows[0].split(',');
rows.shift();
}
rows.forEach(row => {
if(!row) return;
const obj = {};
row.split(',').forEach((val, i) => {
obj[this._headers[i]] = val;
});
this.push(JSON.stringify(obj)+'\n');
});
callback();
}
}
// 使用示例
fs.createReadStream('data.csv')
.pipe(new CSV2JSON())
.pipe(fs.createWriteStream('data.jsonl'));
3. 高并发下的水位控制
当同时处理1000个上传请求时，必须做好背压（backpressure）管理。这与市政供水系统的压力调节如出一辙：
const http = require('http');
const zlib = require('zlib');
http.createServer((req, res) => {
req
.pipe(zlib.createGzip()) // 压缩层
.pipe(encryptStream()) // 加密层
.pipe(uploadToS3()) // 云存储层
.on('error', handleError); // 统一错误处理
// 显示实时上传进度
req.on('data', chunk => {
console.log(`已接收 ${chunk.length} 字节`);
});
}).listen(3000);
4. 流处理进阶技巧
4.1 并发管道焊接
当需要同时进行加密和压缩时，pipeline方法比传统pipe更安全：
const { pipeline } = require('stream');
pipeline(
fs.createReadStream('input.mov'),
zlib.createGzip(),
fs.createWriteStream('output.gz'),
(err) => {
if(err) console.error('处理失败:', err);
else console.log('处理成功');
}
);
4.2 自定义流量阀门
实现限制带宽的自定义流，这在视频点播服务中非常实用：
const { Transform } = require('stream');
class ThrottleStream extends Transform {
constructor(rate) {
super();
this.rate = rate; // KB/s
this.start = Date.now();
this.transferred = 0;
}
_transform(chunk, enc, cb) {
this.transferred += chunk.length;
const elapsed = Date.now() - this.start;
const expected = (this.transferred / 1024) / this.rate * 1000;
const delay = expected - elapsed;
setTimeout(() => {
this.push(chunk);
cb();
}, delay > 0 ? delay : 0);
}
}
// 限制每秒100KB
readStream.pipe(new ThrottleStream(100));
5. 应用场景实战分析
5.1 百万级日志处理
某金融系统需要实时分析交易日志，我们构建的处理管道：
日志文件 -> 解压流 -> 解码流 -> 风控分析流 -> 告警流
↘ 统计流 -> 日报生成流
这种架构使系统吞吐量达到每分钟处理20万条日志。
5.2 视频直播转码
使用FFmpeg的流式接口实现实时转码：
const { spawn } = require('child_process');
const ffmpeg = spawn('ffmpeg', [
'-i', 'pipe:0', // 输入来自标准输入
'-f', 'mp4',
'-vcodec', 'libx264',
'-acodec', 'aac',
'pipe:1' // 输出到标准输出
]);
liveStream.pipe(ffmpeg.stdin);
ffmpeg.stdout.pipe(cdnUploadStream);
6. 技术优缺点深析
优势领域：
- 内存效率：处理10GB文件仅需几十MB内存
- 实时响应：首字节处理时间可控制在毫秒级
- 管道组合：功能模块像乐高积木自由拼接
短板之痛：
- 错误处理犹如高压水枪，稍有不慎就会喷溅
- 调试难度相当于排查漏水的水管网络
- 二进制处理需要精准的Buffer操作技巧
7. 工程师的生存指南
- 背压处理：就像水库大坝，必须有泄洪道
- 内存泄漏预防：确保所有流都有终点管道
- 超时控制：为每个流设置合理的TTL
- 错误传播：管道中的异常需要逐级上报
// 典型的防御式编程示例
function safePipeline(...streams) {
return new Promise((resolve, reject) => {
streams.forEach(s => {
s.on('error', reject);
});
pipeline(...streams, (err) => {
err ? reject(err) : resolve();
});
});
}
8. 实战经验总结
在构建某物联网平台时，我们通过流处理实现了日均处理20TB传感器数据的目标。三个关键教训：
- 水流要有源也要有汇（必须明确流的终点）
- 不同水流间需要缓冲池（合理使用转换流）
- 定期清理河道淤积（及时销毁无用流）