Node.js Stream (流)

在 Node.js 中，Stream 是一个非常重要的概念，用于处理大量数据，特别是那些不适合一次性加载到内存中的数据。Stream 模块提供了一种高效的方式来处理文件、网络连接和其他 I/O 操作中的数据。

Stream 的类型

Node.js 中主要有四种类型的 Stream：

1. Readable Streams (可读流)：可以从中读取数据。
2. Writable Streams (可写流)：可以向其中写入数据。
3. Duplex Streams (双工流)：同时具备可读和可写的能力。
4. Transform Streams (转换流)：继承自 Duplex Streams，可以读取数据并转换后写入。

Stream 的基本概念

• 事件：Stream 通过事件来通知数据的可用性或处理状态。

  • data 事件：每当有数据可读时触发。
  • end 事件：当没有更多数据可读时触发。
  • error 事件：当流中发生错误时触发。
  • finish 事件：当所有数据都被写入并且流被关闭时触发。

• 流的流动模式：Stream 支持两种流动模式：

  • 流模式 (flowing mode)：数据会自动推送。
  • 暂停模式 (paused mode)：数据需要手动读取。

创建和使用 Stream

下面是一些创建和使用 Stream 的基本示例。

1. Readable Streams

1.1 从文件读取数据

使用 fs 模块创建一个可读流。

const fs = require('fs');
const readable = fs.createReadStream('./input.txt');

readable.on('data', (chunk) => {
  console.log(chunk.toString());
});

readable.on('end', () => {
  console.log('Finished reading file.');
});

readable.on('error', (err) => {
  console.error('Error reading file:', err);
});
1
2
3
4
5
6
7
8
9
10
11
12
13
14

2. Writable Streams

2.1 写入文件

使用 fs 模块创建一个可写流。

const fs = require('fs');
const writable = fs.createWriteStream('./output.txt');

writable.write('Hello, ');
writable.write('World!\n');

writable.end('This is the end.');

writable.on('finish', () => {
  console.log('Finished writing to file.');
});

writable.on('error', (err) => {
  console.error('Error writing to file:', err);
});
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

3. Duplex Streams

3.1 使用 net 模块创建 TCP 服务器

TCP 服务器既可以读取数据也可以写入数据。

const net = require('net');

const server = net.createServer((socket) => {
  socket.write('Echo server\r\n');

  socket.on('data', (data) => {
    console.log('Received:', data.toString());
    socket.write(data);
  });

  socket.on('close', () => {
    console.log('Connection closed');
  });
});

server.listen(8080, () => {
  console.log('Server listening on port 8080');
});
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

4. Transform Streams

4.1 使用 Transform Stream 进行数据转换

创建一个简单的转换流，将大写字母转换为小写字母。

const Transform = require('stream').Transform;
const util = require('util');

class UppercaseToLowerCase extends Transform {
  _transform(chunk, encoding, callback) {
    this.push(chunk.toString().toLowerCase());
    callback();
  }
}

const uppercaseToLowerCase = new UppercaseToLowerCase({ objectMode: true });

uppercaseToLowerCase.on('data', (chunk) => {
  console.log(chunk.toString());
});

uppercaseToLowerCase.write('HELLO');
uppercaseToLowerCase.write('WORLD');
uppercaseToLowerCase.end();
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

管道操作

管道操作是 Stream 的一个重要特性，它允许你将一个可读流的数据自动发送到一个可写流。

const fs = require('fs');

const readStream = fs.createReadStream('./input.txt');
const writeStream = fs.createWriteStream('./output.txt');

readStream.pipe(writeStream);

readStream.on('end', () => {
  console.log('Finished piping data.');
});

readStream.on('error', (err) => {
  console.error('Error reading file:', err);
});

writeStream.on('error', (err) => {
  console.error('Error writing to file:', err);
});
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

总结

Stream 是 Node.js 中处理大量数据的关键组件，通过使用 Stream，你可以有效地处理文件读写、网络通信等操作，而不必担心内存限制或性能问题。通过结合使用不同类型的 Stream 和管道操作，你可以构建出高效的数据处理管道。