Node核心模块 之 Stream流

引言

流可以看成数据块，比如要处理一个文件，我们首先需要将文件读入到内存中，如果这个文件很大，那么读入时间就会很长。流可以解决这类问题，使用流可以一次读取一点，直到读完为止，例如以下实例：

// 方案A
var http = require('http');
var fs = require('fs');

var server = http.createServer(function (req, res) {
    fs.readFile(__dirname + '/data.txt', function (err, data) {
        res.end(data);
    });
});
server.listen(8000);

/// 方案B
var http = require('http');
var fs = require('fs');

var server = http.createServer(function (req, res) {
    var stream = fs.createReadStream(__dirname + '/data.txt');
    stream.pipe(res);
});
server.listen(8000);

• 方案A：在每次请求时，都会把整个data.txt文件读入到内存中，然后再把结果返回给客户端。想想看，如果data.txt文件非常大，在响应大量用户的并发请求时，程序可能会消耗大量的内存，这样很可能会造成用户连接缓慢的问题。
• 方案B：(req,res)参数都是流对象，.pipe()方法会自动帮助我们监听data和end事件。上面的这段代码不仅简洁，而且data.txt文件中每一小段数据都将源源不断的发送到客户端。
  除此之外，使用.pipe()方法还有别的好处，比如说它可以自动控制后端压力，以便在客户端连接缓慢的时候node可以将尽可能少的缓存放到内存中。

流模块基础

在node中，一共有五种类型的流：readable,writable,transform,duplex以及"classic"

• readable流
  Readable流可以产出数据，你可以将这些数据传送到一个writable，transform或者duplex流中，只需要调用pipe()方法，在此之前可以用.push方法往readable里放数据
• writable 流
  一个writable流指的是只能流进不能流出的流:

src.pipe(writableStream)

只需要定义一个._write(chunk,enc,next)函数，你就可以将一个readable流的数据释放到其中。chunk代表写进来的数据，enc代表编码的字符串，但是只有在opts.decodeString为false的时候你才可以写一个字符串，next(err)是一个回调函数，使用这个回调函数你可以告诉数据消耗者可以写更多的数据。你可以有选择性的传递一个错误对象error，这时会在流实体上触发一个emit事件

• transform流
  一个流的中间部分，它可以读也可写，但是并不保存数据，它只负责处理流经它的数据，且输入输出是有关联的，如 gzip 模块

var fs = require('fs');
var zlib = require('zlib');

var gzip = zlib.createGzip(); // 返回 transform 流

var inFile = fs.createReadStream('./extra/fileForCompress.txt');
var out = fs.createWriteStream('./extra/fileForCompress.txt.gz');

inFile.pipe(gzip).pipe(out);

• duplex流
  Duplex流是一个可读也可写的流，就好像一个电话，可以接收也可以发送语音。同时保存数据，如 net.Socket 实例

// 服务端
var net = require('net');
var opt = {
    host: '127.0.0.1',
    port: '3000'
};

var server = net.createServer((socket) => {
    socket.on('data', (data) => {
        console.log('client send message: ', data.toString());
    });
    socket.write('hello client');
});
server.listen(opt.port, opt.host, ()=>{
    console.log(server.address());
});

// 客户端
var net = require('net');
var opt = {
    host: '127.0.0.1',
    port: '3000'
};

var client = net.connect(opt, function(){
    client.write('msg from client');  // 可写
});

// 可读
client.on('data', function(data){
    console.log('client: got reply from server [%s]', data);
    client.end();
})

• classic流
  无论何时，只要一个流对象注册了一个data监听器，它就会自动的切换到classic模式，并且根据旧API的方式运行

流事件

• readable可监听 data,end,error等事件，用stream.on(' ',function(){}）

// 消费可读流的常用两种方式：pipe 和 监听data事件
const read = fs.createReadSream('./text.text');
// pipe
read.pipe(process.stdout)
// 监听data事件
read.on('data',function(chunk){
      console.log(chunk)
})
read.on('end',function(){
      console.log('end')
})

• writable

const write = fs.createWriteStream('./a.txt');
write.write('abc');
write.end()

参考：

1、stream-handbook的完整中文版本
2、Stream