Stream的基本概念

Stream的基本概念

Stream是Node.js中处理流式数据的抽象接口，它允许数据以连续流的形式被处理，而不是一次性加载到内存中。这种机制特别适合处理大文件或实时数据，能够显著提高内存效率和程序性能。Node.js中的Stream模块提供了多种类型的流，包括可读流、可写流、双工流和转换流，每种流都有其特定的用途和实现方式。

流的类型

Node.js中的流主要分为四种类型：

1. 可读流（Readable Stream）：用于读取数据，例如从文件中读取内容或从HTTP请求中接收数据。
2. 可写流（Writable Stream）：用于写入数据，例如向文件中写入内容或向HTTP响应中发送数据。
3. 双工流（Duplex Stream）：既可读又可写，例如TCP套接字。
4. 转换流（Transform Stream）：一种特殊的双工流，可以在数据写入和读取时对数据进行转换，例如压缩或加密数据。

可读流

可读流用于从数据源读取数据。它有两种读取模式：流动模式（flowing mode）和暂停模式（paused mode）。在流动模式下，数据会自动从底层系统读取并通过事件触发；在暂停模式下，必须显式调用read()方法才能读取数据。

const fs = require('fs');

// 创建一个可读流
const readableStream = fs.createReadStream('example.txt', 'utf8');

// 监听'data'事件，进入流动模式
readableStream.on('data', (chunk) => {
  console.log('Received chunk:', chunk);
});

// 监听'end'事件
readableStream.on('end', () => {
  console.log('No more data to read.');
});

// 监听'error'事件
readableStream.on('error', (err) => {
  console.error('Error:', err);
});

可写流

可写流用于向目标写入数据。可以通过write()方法写入数据，并通过end()方法标记流的结束。

const fs = require('fs');

// 创建一个可写流
const writableStream = fs.createWriteStream('output.txt');

// 写入数据
writableStream.write('Hello, ');
writableStream.write('World!');

// 标记流结束
writableStream.end();

// 监听'finish'事件
writableStream.on('finish', () => {
  console.log('Data has been written to the file.');
});

// 监听'error'事件
writableStream.on('error', (err) => {
  console.error('Error:', err);
});

双工流

双工流同时实现了可读流和可写流的接口，可以同时进行读写操作。典型的例子是TCP套接字。

const { Duplex } = require('stream');

// 自定义双工流
const myDuplex = new Duplex({
  write(chunk, encoding, callback) {
    console.log('Received:', chunk.toString());
    callback();
  },
  read(size) {
    this.push('Hello from duplex!\n');
    this.push(null); // 表示数据结束
  }
});

// 使用双工流
myDuplex.pipe(process.stdout);
myDuplex.write('Data to duplex stream');

转换流

转换流是一种特殊的双工流，可以在数据写入和读取时对数据进行转换。常见的例子是压缩或加密数据。

const { Transform } = require('stream');

// 自定义转换流，将输入转换为大写
const upperCaseTransform = new Transform({
  transform(chunk, encoding, callback) {
    this.push(chunk.toString().toUpperCase());
    callback();
  }
});

// 使用转换流
process.stdin.pipe(upperCaseTransform).pipe(process.stdout);

流的管道

管道（pipe）是将多个流连接起来的便捷方式，可以将可读流的数据直接传输到可写流中，无需手动处理数据事件。

const fs = require('fs');

// 创建可读流和可写流
const readableStream = fs.createReadStream('input.txt');
const writableStream = fs.createWriteStream('output.txt');

// 使用管道连接流
readableStream.pipe(writableStream);

// 监听完成事件
writableStream.on('finish', () => {
  console.log('Data has been piped successfully.');
});

错误处理

流操作中可能会发生错误，因此需要妥善处理错误事件，避免程序崩溃。

const fs = require('fs');

const readableStream = fs.createReadStream('nonexistent.txt');
const writableStream = fs.createWriteStream('output.txt');

readableStream.on('error', (err) => {
  console.error('Read error:', err);
});

writableStream.on('error', (err) => {
  console.error('Write error:', err);
});

readableStream.pipe(writableStream);

流的性能优势

流的主要优势在于其内存效率。通过分块处理数据，流可以避免一次性加载大量数据到内存中，特别适合处理大文件或实时数据流。

const fs = require('fs');

// 不使用流的方式读取大文件（内存消耗高）
fs.readFile('largefile.txt', (err, data) => {
  if (err) throw err;
  console.log('File size:', data.length);
});

// 使用流的方式读取大文件（内存效率高）
const stream = fs.createReadStream('largefile.txt');
let size = 0;

stream.on('data', (chunk) => {
  size += chunk.length;
});

stream.on('end', () => {
  console.log('File size:', size);
});

自定义流

除了使用内置的流类型，还可以通过继承Readable、Writable、Duplex或Transform类来自定义流。

const { Readable } = require('stream');

// 自定义可读流
class MyReadable extends Readable {
  constructor(options) {
    super(options);
    this.data = ['Hello', 'World', '!'];
    this.index = 0;
  }

  _read(size) {
    if (this.index < this.data.length) {
      this.push(this.data[this.index]);
      this.index++;
    } else {
      this.push(null); // 结束流
    }
  }
}

const myReadable = new MyReadable();
myReadable.on('data', (chunk) => {
  console.log(chunk.toString());
});

流的背压机制

背压（backpressure）是流处理中的重要概念，用于控制数据流速，避免生产者（可读流）速度过快导致消费者（可写流）无法及时处理。

const fs = require('fs');

const readableStream = fs.createReadStream('largefile.txt');
const writableStream = fs.createWriteStream('output.txt');

// 手动处理背压
readableStream.on('data', (chunk) => {
  const canContinue = writableStream.write(chunk);
  if (!canContinue) {
    readableStream.pause(); // 暂停读取
    writableStream.once('drain', () => {
      readableStream.resume(); // 恢复读取
    });
  }
});

readableStream.on('end', () => {
  writableStream.end();
});

流的事件

流是EventEmitter的实例，可以触发和监听多种事件，包括data、end、error、finish等。

const fs = require('fs');

const readableStream = fs.createReadStream('example.txt');

// 监听各种事件
readableStream.on('open', () => {
  console.log('File opened');
});

readableStream.on('ready', () => {
  console.log('File ready');
});

readableStream.on('close', () => {
  console.log('File closed');
});

流的实际应用

流在实际开发中有广泛的应用场景，例如文件处理、网络通信、数据转换等。

const http = require('http');
const fs = require('fs');

// 使用流处理HTTP请求和文件
http.createServer((req, res) => {
  const fileStream = fs.createReadStream('largefile.txt');
  fileStream.pipe(res); // 将文件内容直接传输到HTTP响应
}).listen(3000);