nodejs系列（9）stream (流)

港控/mmm° 2021-09-14 16:06 307阅读 0赞

流（stream）在 Node.js 中是处理流数据的抽象接口（abstract interface）。 stream 模块提供了基础的 API 。使用这些 API 可以很容易地来构建实现流接口的对象。Node.js 提供了多种流对象。 例如， HTTP 请求 和 process.stdout 就都是流的实例。流可以是可读的、可写的，或是可读写的。**所有的流都是 EventEmitter 的实例**。

# 流的类型                                                                      #

流是有方向的，就像水管一样，一根接一根，程序上用方法pipe（）连接起来，而水管里面的水就是流数据，流 数据格式通常为Buffer数据格式。流的类型可以分为4种：

设备流向程序：Readable - 可读的流 (例如 fs.createReadStream()).  
程序流向设备：Writable - 可写的流 (例如 fs.createWriteStream()).  
双向：Duplex - 可读写的流 (例如 net.Socket).

双向：Transform - 在读写过程中可以修改和变换数据的 Duplex 流 (例如 zlib.createDeflate()).

# 存储格式                                                                      #

Writable 和 Readable 流都会将数据存储到内部的缓冲器（buffer）中， 要获取这些buffer数据，可以通过writable.\_writableState.getBuffer() 或 readable.\_readableState.buffer（writable和readable是指可写流和可读流的实例对象）实现。其中缓冲器的大小取决于传递给流构造函数的 highWaterMark 选项。

# 存储过程                                                                      #

无论是读或写都有一个阈值，或叫极值。当可读流的实现调用 stream.push(chunk) 方法时，数据被放到缓冲器中。如果流的消费者 没有调用 stream.read() 方法， 这些数据会始终存在于内部队列中，直到被消费。

在读数据过程中，当内部可读缓冲器的大小达到 highWaterMark 指定的阈值时，流会暂停从底层资源读取数据，直到当前 缓冲器的数据被消费 (也就是说， 流会在内部停止调用 readable.\_read() 来填充可读缓冲器)。

在写数据过程中，可写流通过反复调用 writable.write(chunk) 方法将数据放到缓冲器。 当内部可写缓冲器的总大小小于 highWaterMark 指定的阈值时， 调用 writable.write() 将返回true。 一旦内部缓冲器的大小达到或超过 highWaterMark ，调用 writable.write() 将返回 false 。

stream API 的关键目标， 尤其对于 stream.pipe() 方法， 就是限制缓冲器数据大小，以达到可接受的程度。这样，对于读写速度不匹配的源头和目标，就不会超出可用的内存大小。

# 双向流                                                                         #

双向流Duplex 和 Transform 都是可读写的。 在内部，它们都维护了两个相互独立的缓冲器用于读和写。 在维持了合理高效的数据流的同时，也使得对于读和写可以独立进行而互不影响。 例如， net.Socket 就是 Duplex 的实例，它的可读端可以消费从套接字（socket）中接收的数据， 可写端则可以将数据写入到套接字。 由于数据写入到套接字中的速度可能比从套接字接收数据的速度快或者慢， 在读写两端使用独立缓冲器，并进行独立操作就显得很重要了。这个在本篇不做详细介绍，在后续博文中会有专门介绍，本篇博客主要介绍读和写。

# 流读取数据                                                                  #

从流中读取数据 ，读取的监听事件有data, end, 和 error，读取时可设定具体的开始和结束的读取位置，默认 从0开始读取。读取的方法为fs.createReadStream（如fs.createReadStream('sample.txt', \{ start: 90, end: 99 \});）。与方法fs.readFile（如fs.readFile('/etc/passwd', 'utf8', callback);） 的区别在于fs.createReadStream就像水流一样从设备中一段段读取数据分段传输给程序中，而fs.readFile则是一次性地读取所有数据然后传输给程序，当遇到大文件时，不建议使用fs.readFile方法。

# 流写入数据                                                                  #

通过流写入数据，写入的监听事件有finish和error，创建写入流的方法为fs.createWriteStream，该方法可带一个配置参数对象，该配置对象最常用的可选参数有3种选项start、encoding 和flags。start选项，使其可以写入数据到文件某个位置。 如果是修改一个文件而不是覆盖它，则需要flags 模式为 r+ 而不是默认的 w 模式。 encoding 可以是任何可以被 Buffer 接受的值。同理于读取流，与fs.writeFile的区别在于是分段写入还是一次性写入设备。

//可选参数
    var defaults = {
    	flags: 'w',
    	encoding: 'utf8',
    	start : 0
    };
    
    var fs = require("fs");
    var txt = '大家好，我是output文本。';
    var writerStream = fs.createWriteStream('output.txt', defaults); // 创建一个可以写入的流，写入到文件 output.txt 中
    writerStream.write(txt, 'UTF8'); // 使用 utf8 编码写入数据
    writerStream.end(); // 标记文件末尾
    writerStream.on('finish', function() {
    	console.log("写入完成。");
    });
    writerStream.on('error', function(err) {
    	console.log(err.stack);
    });

# 管道流                                                                         #

管道流是指一个输出流到输入流的机制：设备->程序->设备 ，该过程通过pipe( )方法连接两个设备进行数据传输。简单来说，就如同普通的复制功能。

var fs = require("fs");
    var readerStream = fs.createReadStream('input.txt'); // 创建一个可读流
    var writerStream = fs.createWriteStream('output.txt'); // 创建一个可写流
    console.log(writerStream);
    readerStream.pipe(writerStream); // 管道读写操作，读取 input.txt 文件内容，并将内容写入到 output.txt 文件中

# 链式流                                                                         #

链式流与管道流类似，是指通过连接输出流到另外一个流并创建多个对个流操作链的机制，即中间多了一步流数据处理的步骤：设备->程序->流数据处理->设备。要使用链式流需先导入through2模块。导入命令如下：

cnpm install through2

如将input.txt所有字母转变为大写并写入output.txt：

var fs = require("fs");
    var readerStream = fs.createReadStream('input.txt'); // 创建一个可读流
    var writerStream = fs.createWriteStream('output.txt'); // 创建一个可写流
    var through2 = require('through2');
    var upper = function() {
    	var result = through2(function(buf, _, next) {
    		this.push(buf.toString().toUpperCase());
    		next();
    	})
    	return result;
    }
    readerStream.pipe(upper()).pipe(writerStream);

# 从流到函数                                                                  #

有时候流读取文件并不一定要有输出文件，也可以把读取的数据进行函数操作，如数据分析或写入数据库等。实现该功能前需引入concat-stream，下面为一个实现数据倒置但没有输入文件的实例：

var fs = require('fs');
    var readerStream = fs.createReadStream('input.txt'); // 创建一个可读流
    var writerStream = fs.createWriteStream('output.txt'); // 创建一个可写流
    var concat = require('concat-stream');
    var reverseStream = concat(function(text) {
    	console.log(text.toString().split("").reverse().join(""));
    })
    readerStream.pipe(reverseStream);

# 手动控制流读写：大文件复制                                           #

文件进行流的读写有时候会遇到阈值，如读取的速度比写入的速度要快时，若继续读取，在写入时有可能造成数据丢失，此时可以通过stream.pause( )和stream.resume( )来手动控制。如大文件复制实例如下：

（注）'drain' 事件：如果调用 stream.write(chunk) 方法返回 false，流将在适当的时机触发 'drain' 事件，这时才可以继续向流中写入数据。

# 流的在请求中的实际用法                                               #

HTTP请求中的request对象和response对象都是流对象，所以有如下写法来和客户端交互：

var http = require('http');
    var fs = require('fs');
    
    var server = http.createServer(function (req, res) {
        let stream = fs.createReadStream(__dirname + '/data.txt');//创造可读流
        stream.pipe(res);//将可读流写入response
    });
    server.listen(8000);