Node.js 中的流完整指南

Node.js 中的流是一种以连续方式处理数据读写的强大方法。它们使您能够有效地处理数据，特别是在处理大量信息或 I/O 操作时。本指南将介绍流的类型、如何使用它们以及实际示例，以帮助您了解流在 Node.js 中的工作原理。

什么是流？

流是允许您以连续方式从源读取数据或将数据写入目标的对象。它们非常适合逐个处理数据，而不是一次读取或写入整个文件或缓冲区。这在处理大型数据集时特别有用，因为它可以显着减少内存使用量。

关键概念

• 流动模式：数据自动从源流向目的地。
• 暂停模式：需要手动从源读取数据。
• 可读流：可以从中读取数据的流。
• 可写流：可以写入数据的流。
• 双工流：既可以读取也可以写入数据的流。
• 转换流：在读取或写入数据时修改或转换数据的流。

流的类型

1. 可读流：这些流允许您读取数据。示例包括 fs.createReadStream() 和 http.IncomingMessage.

2. 可写流：这些流允许您写入数据。示例包括 fs.createWriteStream() 和 http.ServerResponse.

3. 双工流：这些流可以读取和写入数据。示例包括 TCP 套接字和 net.Duplex.

4. 转换流：这是一种双工流，可以在读取或写入数据时修改数据。示例包括用于压缩的 zlib.createGzip()。

创建可读流

您可以使用内置的 fs 模块创建可读流来读取文件，或者使用stream.Readable创建自定义可读流。

示例：使用可读流读取文件

const fs = require('fs');

// Create a readable stream
const readableStream = fs.createReadStream('example.txt', { encoding: 'utf8' });

// Handling the 'data' event
readableStream.on('data', (chunk) => {
    console.log('New chunk received:', chunk);
});

// Handling the 'end' event
readableStream.on('end', () => {
    console.log('No more data to read.');
});

示例：自定义可读流

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

class MyReadableStream extends Readable {
    constructor(options) {
        super(options);
        this.current = 0;
    }

    _read(size) {
        if (this.current  {
    console.log('Received:', chunk.toString());
});

创建可写流

您可以使用 fs 模块或扩展stream.Writable 类来创建可写流。

示例：使用可写流写入文件

const fs = require('fs');

// Create a writable stream
const writableStream = fs.createWriteStream('output.txt');

// Write data to the stream
writableStream.write('Hello, World!\n');
writableStream.write('Writing to a file using streams.\n');

// End the stream
writableStream.end(() => {
    console.log('Finished writing to file.');
});

示例：自定义可写流

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

class MyWritableStream extends Writable {
    _write(chunk, encoding, callback) {
        console.log('Writing:', chunk.toString());
        callback(); // Call when done
    }
}

const myWritableStream = new MyWritableStream();
myWritableStream.write('Hello, World!\n');
myWritableStream.write('Writing to custom writable stream.\n');
myWritableStream.end();

使用双工流

双工流可以同时读取和写入数据。一个常见的用例是 TCP 套接字。

示例：创建双工流

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

class MyDuplexStream extends Duplex {
    _read(size) {
        this.push('Data from duplex stream\n');
        this.push(null); // No more data
    }

    _write(chunk, encoding, callback) {
        console.log('Received:', chunk.toString());
        callback();
    }
}

const myDuplexStream = new MyDuplexStream();
myDuplexStream.on('data', (chunk) => {
    console.log('Reading:', chunk.toString());
});

// Write to the duplex stream
myDuplexStream.write('Hello, Duplex!\n');
myDuplexStream.end();

使用转换流

转换流对于修改流经流的数据非常有用。例如，您可以使用转换流来压缩数据。

示例：创建转换流

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

class MyTransformStream extends Transform {
    _transform(chunk, encoding, callback) {
        const upperChunk = chunk.toString().toUpperCase();
        this.push(upperChunk);
        callback();
    }
}

const myTransformStream = new MyTransformStream();
myTransformStream.on('data', (chunk) => {
    console.log('Transformed:', chunk.toString());
});

// Pipe data through the transform stream
process.stdin.pipe(myTransformStream).pipe(process.stdout);

管道流

流的强大功能之一是将它们通过管道连接在一起的能力。管道允许您将可读流连接到可写流，这使得传输数据变得容易。

示例：管道流

const fs = require('fs');

// Create a readable stream
const readableStream = fs.createReadStream('input.txt');

// Create a writable stream
const writableStream = fs.createWriteStream('output.txt');

// Pipe the readable stream to the writable stream
readableStream.pipe(writableStream);

writableStream.on('finish', () => {
    console.log('Data has been written to output.txt');
});

Node.js 中的流事件

1. 可读流事件

可读流发出几个重要事件来帮助您管理数据流：

• data：当一大块数据可供读取时发出。
• end：当没有更多数据可供读取时发出。
• error：读取过程中发生错误时发出。
• close：当流和任何底层资源（如文件描述符）已关闭时发出。

示例：可读流事件

const fs = require('fs');

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

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

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

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

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

2. 可写流事件

可写流也会发出几个事件：

• drain：当流已满后准备好接受更多数据时发出。
• finish：当所有数据都写入流并且调用了 end() 方法时发出。
• error：写入过程中发生错误时发出。
• close：当流和任何底层资源关闭时发出。

示例：可写流事件

const fs = require('fs');

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

writableStream.on('finish', () => {
    console.log('All data has been written to output.txt');
});

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

// Writing data
writableStream.write('Hello, World!\n');
writableStream.write('Writing to a file using streams.\n');
writableStream.end(); // Call end to finish the writing process

3. 转换流事件

转换流从可读流和可写流继承事件，并且它们发出：

• data：当转换后的块可供读取时发出。
• end：当没有更多数据可供转换时发出。
• error：转换过程中发生错误时发出。
• finish：当所有数据都已处理并写入时发出。

示例：转换流事件

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

class MyTransformStream extends Transform {
    _transform(chunk, encoding, callback) {
        const upperChunk = chunk.toString().toUpperCase();
        this.push(upperChunk);
        callback();
    }
}

const myTransformStream = new MyTransformStream();

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

myTransformStream.on('end', () => {
    console.log('No more data to transform.');
});

myTransformStream.on('error', (err) => {
    console.error('Error occurred:', err);
});

// Write data to the transform stream
myTransformStream.write('Hello, World!\n');
myTransformStream.write('Transforming this text.\n');
myTransformStream.end(); // End the stream

活动概要

• 可读流：数据、结束、错误、关闭
• 可写流：排出、完成、错误、关闭
• Transform Streams：从可读流和可写流继承数据、结束、错误、完成

结论

Node.js 中的流提供了一种强大且高效的方式来连续处理数据。它们允许您逐段读取和写入数据，这使得它们对于 I/O 操作和处理大型数据集特别有用。了解如何创建和使用不同类型的流，以及如何处理事件，将帮助您在 Node.js 中构建更高效、可扩展的应用程序。

无论您是创建可读、可写、双工还是转换流，流 API 的灵活性都允许您以最适合您的应用程序需求的方式处理数据。