Node.js Streams 与背压机制：处理大文件的正确姿势

1
import { createReadStream } from 'fs';
2
import { Readable } from 'stream';
3

4
// 从文件创建可读流
5
const fileStream = createReadStream('/var/log/app.log', {
6
  encoding: 'utf-8',
7
  highWaterMark: 64 * 1024 // 每次读取 64KB
8
});
9

10
// 事件模式（flowing mode）
11
fileStream.on('data', (chunk) => {
12
  console.log(`Received ${chunk.length} bytes`);
13
});
14

15
fileStream.on('end', () => {
16
  console.log('File reading complete');
17
});
18

19
fileStream.on('error', (err) => {
20
  console.error('Read error:', err);
21
});

1
import { Transform } from 'stream';
2

3
// CSV 转 JSON 转换流
4
class CSVToJSON extends Transform {
5
  #headers = null;
6
  #buffer = '';
7

8
  constructor(options = {}) {
9
    super({ ...options, readableObjectMode: true });
10
  }
11

12
  _transform(chunk, encoding, callback) {
13
    this.#buffer += chunk.toString();
14
    const lines = this.#buffer.split('\n');
15
    this.#buffer = lines.pop() || ''; // 保留不完整的行
16

17
    for (const line of lines) {
18
      if (!line.trim()) continue;
19

20
      const values = line.split(',').map(v => v.trim());
21

22
      if (!this.#headers) {
23
        this.#headers = values;
24
        continue;
25
      }
26

27
      const obj = {};
28
      this.#headers.forEach((header, i) => {
29
        obj[header] = values[i];
30
      });
31

32
      this.push(obj);
33
    }
34

35
    callback();
36
  }
37

38
  _flush(callback) {
39
    // 处理最后残留的数据
40
    if (this.#buffer.trim() && this.#headers) {
41
      const values = this.#buffer.split(',').map(v => v.trim());
42
      const obj = {};
43
      this.#headers.forEach((header, i) => {
44
        obj[header] = values[i];
45
      });
46
      this.push(obj);
47
    }
48
    callback();
49
  }
50
}
51

52
// 使用
53
import { createReadStream } from 'fs';
54

55
const csvStream = createReadStream('./data.csv');
56
const parser = new CSVToJSON();
57

58
csvStream.pipe(parser).on('data', (record) => {
59
  console.log(record);
60
  // { name: 'Alice', age: '30', email: 'alice@test.com' }
61
});

1
import { Duplex } from 'stream';
2

3
class EchoStream extends Duplex {
4
  #queue = [];
5

6
  _write(chunk, encoding, callback) {
7
    // 写入端接收数据，转换后放入队列
8
    this.#queue.push(chunk.toString().toUpperCase());
9
    callback();
10
  }
11

12
  _read(size) {
13
    if (this.#queue.length > 0) {
14
      this.push(this.#queue.shift());
15
    } else {
16
      // 没有数据可读时，延迟重试
17
      setTimeout(() => this._read(size), 10);
18
    }
19
  }
20
}

1
// ✅ 正确做法：手动处理背压
2
readable.on('data', (chunk) => {
3
  const canContinue = writable.write(chunk);
4

5
  if (!canContinue) {
6
    // 内部缓冲区满了，暂停读取
7
    readable.pause();
8

9
    // 等缓冲区排空后再继续
10
    writable.once('drain', () => {
11
      readable.resume();
12
    });
13
  }
14
});
15

16
readable.on('end', () => {
17
  writable.end();
18
});

1
生产者 (Readable)          消费者 (Writable)
2
  │                           │
3
  │  chunk ──────────────►  内部缓冲区
4
  │  chunk ──────────────►  [####............]  ← highWaterMark
5
  │  chunk ──────────────►  [########........]
6
  │  chunk ──────────────►  [############....]
7
  │  chunk ──────────────►  [################]  ← 缓冲区满！
8
  │                           │
9
  │  write() returns false    │
10
  │  ← 暂停生产               │
11
  │                           │  → 慢慢消费...
12
  │                           │  [########....]
13
  │                           │  [............]  ← 缓冲区清空
14
  │                           │
15
  │  'drain' event ─────────► │
16
  │  → 恢复生产               │

1
// ❌ pipe 的问题：
2
// 1. 不会自动传播错误
3
// 2. 错误时不会自动清理流
4
// 3. 需要手动给每个流添加 error handler
5
readable
6
  .pipe(transform1)
7
  .pipe(transform2)
8
  .pipe(writable);
9
// 如果 transform1 出错，readable 和 writable 不会被清理
10
// → 文件描述符泄漏
11

12
// ✅ pipeline 解决所有问题
13
import { pipeline } from 'stream';
14

15
pipeline(
16
  readable,
17
  transform1,
18
  transform2,
19
  writable,
20
  (err) => {
21
    if (err) {
22
      console.error('Pipeline failed:', err);
23
    } else {
24
      console.log('Pipeline succeeded');
25
    }
26
    // 所有流都已正确清理
27
  }
28
);
29

30
// 或者 Promise 版本
31
import { pipeline as pipelineAsync } from 'stream/promises';
32

33
try {
34
  await pipelineAsync(readable, transform1, transform2, writable);
35
} catch (err) {
36
  console.error('Pipeline failed:', err);
37
}

1
import { createReadStream } from 'fs';
2
import { pipeline } from 'stream/promises';
3
import { Transform, Writable } from 'stream';
4

5
// 行分割
6
class LineParser extends Transform {
7
  #buffer = '';
8

9
  constructor() {
10
    super({ readableObjectMode: true });
11
  }
12

13
  _transform(chunk, encoding, callback) {
14
    this.#buffer += chunk.toString();
15
    const lines = this.#buffer.split('\n');
16
    this.#buffer = lines.pop() || '';
17
    for (const line of lines) {
18
      if (line.trim()) this.push(line);
19
    }
20
    callback();
21
  }
22

23
  _flush(callback) {
24
    if (this.#buffer.trim()) this.push(this.#buffer);
25
    callback();
26
  }
27
}
28

29
// CSV 解析（简化版）
30
class CSVParser extends Transform {
31
  #headers = null;
32

33
  constructor() {
34
    super({ objectMode: true });
35
  }
36

37
  _transform(line, encoding, callback) {
38
    const values = line.split(',').map(v => v.trim().replace(/^"|"$/g, ''));
39

40
    if (!this.#headers) {
41
      this.#headers = values;
42
    } else {
43
      const record = {};
44
      this.#headers.forEach((h, i) => { record[h] = values[i]; });
45
      this.push(record);
46
    }
47
    callback();
48
  }
49
}
50

51
// 统计聚合
52
class Aggregator extends Writable {
53
  stats = {
54
    total: 0,
55
    sumAge: 0,
56
    cities: new Map(),
57
    minAge: Infinity,
58
    maxAge: -Infinity
59
  };
60

61
  constructor() {
62
    super({ objectMode: true });
63
  }
64

65
  _write(record, encoding, callback) {
66
    this.stats.total++;
67

68
    const age = Number(record.age);
69
    if (!isNaN(age)) {
70
      this.stats.sumAge += age;
71
      this.stats.minAge = Math.min(this.stats.minAge, age);
72
      this.stats.maxAge = Math.max(this.stats.maxAge, age);
73
    }
74

75
    const city = record.city || 'Unknown';
76
    this.stats.cities.set(city, (this.stats.cities.get(city) || 0) + 1);
77

78
    callback();
79
  }
80

81
  _final(callback) {
82
    console.log('\n=== Statistics ===');
83
    console.log(`Total records: ${this.stats.total}`);
84
    console.log(`Average age: ${(this.stats.sumAge / this.stats.total).toFixed(1)}`);
85
    console.log(`Age range: ${this.stats.minAge} - ${this.stats.maxAge}`);
86
    console.log('\nTop cities:');
87

88
    const sorted = [...this.stats.cities.entries()]
89
      .sort((a, b) => b[1] - a[1])
90
      .slice(0, 10);
91

92
    for (const [city, count] of sorted) {
93
      console.log(`  ${city}: ${count}`);
94
    }
95

96
    callback();
97
  }
98
}
99

100
// 运行
101
await pipeline(
102
  createReadStream('./users.csv', { highWaterMark: 256 * 1024 }),
103
  new LineParser(),
104
  new CSVParser(),
105
  new Aggregator()
106
);

1
import { createReadStream, createWriteStream, statSync } from 'fs';
2
import { pipeline } from 'stream/promises';
3
import { Transform } from 'stream';
4

5
// JSON 日志解析
6
class JSONLogParser extends Transform {
7
  #buffer = '';
8

9
  constructor() {
10
    super({ readableObjectMode: true });
11
  }
12

13
  _transform(chunk, encoding, callback) {
14
    this.#buffer += chunk.toString();
15
    const lines = this.#buffer.split('\n');
16
    this.#buffer = lines.pop() || '';
17

18
    for (const line of lines) {
19
      try {
20
        const log = JSON.parse(line);
21
        this.push(log);
22
      } catch {
23
        // 跳过无效的 JSON 行
24
      }
25
    }
26
    callback();
27
  }
28

29
  _flush(callback) {
30
    if (this.#buffer.trim()) {
31
      try {
32
        this.push(JSON.parse(this.#buffer));
33
      } catch { /* skip */ }
34
    }
35
    callback();
36
  }
37
}
38

39
// 错误日志过滤
40
class ErrorFilter extends Transform {
41
  constructor() {
42
    super({ objectMode: true });
43
  }
44

45
  _transform(log, encoding, callback) {
46
    if (log.level === 'error' || log.level === 'fatal') {
47
      this.push(log);
48
    }
49
    callback();
50
  }
51
}
52

53
// 格式化输出
54
class ErrorFormatter extends Transform {
55
  constructor() {
56
    super({ objectMode: true, writableObjectMode: true });
57
  }
58

59
  _transform(log, encoding, callback) {
60
    const formatted = [
61
      `[${log.timestamp || new Date().toISOString()}]`,
62
      `[${(log.level || 'UNKNOWN').toUpperCase()}]`,
63
      log.message || 'No message',
64
      log.stack ? `\n  Stack: ${log.stack}` : '',
65
      log.context ? `\n  Context: ${JSON.stringify(log.context)}` : '',
66
      '\n---\n'
67
    ].join(' ');
68

69
    this.push(formatted);
70
    callback();
71
  }
72
}
73

74
// 带进度的处理
75
const fileSize = statSync('./app.log').size;
76

77
class Progress extends Transform {
78
  #processed = 0;
79

80
  constructor(total) {
81
    super();
82
    this.total = total;
83
  }
84

85
  _transform(chunk, encoding, callback) {
86
    this.#processed += chunk.length;
87
    const pct = ((this.#processed / this.total) * 100).toFixed(1);
88
    process.stderr.write(`\rProcessing: ${pct}%`);
89
    this.push(chunk);
90
    callback();
91
  }
92

93
  _flush(callback) {
94
    process.stderr.write('\rProcessing: 100% — Done!\n');
95
    callback();
96
  }
97
}
98

99
await pipeline(
100
  createReadStream('./app.log'),
101
  new Progress(fileSize),
102
  new JSONLogParser(),
103
  new ErrorFilter(),
104
  new ErrorFormatter(),
105
  createWriteStream('./errors.log')
106
);
107

108
console.log('Error extraction complete');

1
import { createServer } from 'http';
2
import { createWriteStream } from 'fs';
3
import { pipeline } from 'stream/promises';
4
import { createGunzip } from 'zlib';
5
import { Transform } from 'stream';
6
import { randomUUID } from 'crypto';
7

8
// 限速转换流
9
class RateLimiter extends Transform {
10
  #bytesPerSecond;
11
  #bytesThisSecond = 0;
12
  #lastReset = Date.now();
13

14
  constructor(bytesPerSecond) {
15
    super();
16
    this.#bytesPerSecond = bytesPerSecond;
17
  }
18

19
  _transform(chunk, encoding, callback) {
20
    const now = Date.now();
21
    if (now - this.#lastReset >= 1000) {
22
      this.#bytesThisSecond = 0;
23
      this.#lastReset = now;
24
    }
25

26
    this.#bytesThisSecond += chunk.length;
27

28
    if (this.#bytesThisSecond > this.#bytesPerSecond) {
29
      const delay = 1000 - (now - this.#lastReset);
30
      setTimeout(() => {
31
        this.push(chunk);
32
        callback();
33
      }, delay);
34
    } else {
35
      this.push(chunk);
36
      callback();
37
    }
38
  }
39
}
40

41
// 大小限制转换流
42
class SizeLimiter extends Transform {
43
  #maxBytes;
44
  #received = 0;
45

46
  constructor(maxBytes) {
47
    super();
48
    this.#maxBytes = maxBytes;
49
  }
50

51
  _transform(chunk, encoding, callback) {
52
    this.#received += chunk.length;
53

54
    if (this.#received > this.#maxBytes) {
55
      callback(new Error(`Upload exceeds maximum size of ${this.#maxBytes} bytes`));
56
      return;
57
    }
58

59
    this.push(chunk);
60
    callback();
61
  }
62
}
63

64
const server = createServer(async (req, res) => {
65
  if (req.method !== 'POST' || req.url !== '/upload') {
66
    res.writeHead(404);
67
    res.end('Not Found');
68
    return;
69
  }
70

71
  const fileId = randomUUID();
72
  const filePath = `./uploads/${fileId}.dat`;
73
  const isGzipped = req.headers['content-encoding'] === 'gzip';
74

75
  const streams = [
76
    req,
77
    new SizeLimiter(100 * 1024 * 1024), // 100MB 限制
78
  ];
79

80
  if (isGzipped) {
81
    streams.push(createGunzip());
82
  }
83

84
  streams.push(createWriteStream(filePath));
85

86
  try {
87
    await pipeline(...streams);
88
    res.writeHead(200, { 'Content-Type': 'application/json' });
89
    res.end(JSON.stringify({ fileId, path: filePath }));
90
  } catch (err) {
91
    res.writeHead(413, { 'Content-Type': 'application/json' });
92
    res.end(JSON.stringify({ error: err.message }));
93
  }
94
});
95

96
server.listen(3000, () => {
97
  console.log('Upload server running on :3000');
98
});

1
import { compose } from 'stream';
2

3
// 将多个 Transform 组合成一个可复用的流
4
const logProcessor = compose(
5
  new LineParser(),
6
  new JSONLogParser(),
7
  new ErrorFilter(),
8
  new ErrorFormatter()
9
);
10

11
// 可以像单个 Transform 一样使用
12
await pipeline(
13
  createReadStream('./app.log'),
14
  logProcessor,
15
  createWriteStream('./errors.log')
16
);

1
import { Transform } from 'stream';
2

3
// 工厂函数创建 Transform
4
function createTransform(transformFn, flushFn = null) {
5
  return new Transform({
6
    objectMode: true,
7
    transform(chunk, encoding, callback) {
8
      try {
9
        const result = transformFn(chunk);
10
        if (result !== undefined && result !== null) {
11
          this.push(result);
12
        }
13
        callback();
14
      } catch (err) {
15
        callback(err);
16
      }
17
    },
18
    flush(callback) {
19
      if (flushFn) {
20
        try {
21
          const result = flushFn();
22
          if (result !== undefined) this.push(result);
23
        } catch (err) {
24
          return callback(err);
25
        }
26
      }
27
      callback();
28
    }
29
  });
30
}
31

32
// 使用
33
const toUpperCase = () => createTransform(s => s.toString().toUpperCase());
34
const addLineNumbers = () => {
35
  let n = 0;
36
  return createTransform(line => `${++n}: ${line}`);
37
};
38
const filterEmpty = () => createTransform(line => line.trim() ? line : undefined);
39

40
await pipeline(
41
  createReadStream('./input.txt'),
42
  new LineParser(),
43
  filterEmpty(),
44
  toUpperCase(),
45
  addLineNumbers(),
46
  createTransform(line => line + '\n'),
47
  createWriteStream('./output.txt')
48
);