node中的流（stream）

官方的定义

流（stream）是 Node.js 中处理流式数据的抽象接口。stream模块用于构建实现了流接口的对象。

咱们通常是直接使用node提供的流对象，例如在服务器请求、文件模块中使用。

流的分类

• Writable- 可写入数据的流（例如fs.createWriteStream()）。
• Readable- 可读取数据的流（例如fs.createReadStream()）。
• Duplex- 可读又可写的流（例如net.Socket）。
• Transform- 在读写过程当中能够修改或转换数据的Duplex流（例如zlib.createDeflate()）

流的重要事件和方法

下面会结合具体的例子来梳理一下流的经常使用事件和方法，加深对流的理解。

一、data和end事件

流类型：可读流 data事件在可读流将数据传给消费者后触发，特别注意的是，添加了该事件的流会自动切换为流动模式， end事件在当流中没有数据可供消费时触发:

const stream = fs.createReadStream('./file.txt') // 默认静止态
let chunks = []
stream.on("data", (chunk) => { // 变成流动态
    chunks.push(chunk)
})
stream.on("end", () => {
    constcontent = Buffer.concat(chunks).toString()
    console.log(content)
})
复制代码

其中chunk是buffer类型 补充，在任意时刻，可读流会处于如下三种状态之一：

• readable.readableFlowing === null
• readable.readableFlowing === false
• readable.readableFlowing === true 初始时则readable.readableFlowing为null，添加data事件后变为true。调用readable.pause()、readable.unpipe()、或接收到背压，则readable.readableFlowing会被设为false，在这个状态下，为data事件绑定监听器不会使readable.readableFlowing切换到 true

二、readable事件与read()方法

流类型：可读流 readable事件代表流有新的动态：要么有新的数据，要么到达流的尽头。下面是读取文件的例子：

const stream = fs.createReadStream('./file.txt')
let chunks = []
// stream中无数据也会触发readable， 此时read方法获得null.
// 读取到流的尽头也会触发，而且在end事件以前
stream.on("readable", () => {
    console.log('触发readable');
    let data;
    while (data = stream.read(1024)) {
        chunks.push(data)
        console.log('读取数据', data);
    }
})
stream.on("end", () => {
    const content = Buffer.concat(chunks).toString();
    console.log(content)
})
复制代码

**使用readable会使流的状态变成暂停模式，即便监听了data事件。在调用read方法且有返回数据时会触发data事件。**上面代码中，read方法读取内部缓冲中的数据，若是不指定size参数，则是读取内部缓冲中的全部数据，注意不是流中的全部数据，不指定size的话也就不必使用while循环了，直接一次性读取，while循环代码块可变为：

data = stream.read()
data && chunks.push(data)
console.log('读取数据', data);
复制代码

根据运行结果，read方法将缓冲区数据读完后会触发readable事件，也就是当read()返回null后触发。 这是第二种读取可读流的模式，即经过read()读取

三、pipe()和unpipe()

流类型：可读流 定义见官方文档，下面例子使用pipe响应http请求

const http = require('http')
const fs = require('fs')
const server = http.createServer()
server.on('request', (request, response) => {
    const stream = fs.createReadStream('./file.txt')
    stream.pipe(response)
})
server.listen(8888)
复制代码

使用pipe时数据流会被自动管理，因此即便可读流更快，目标可写流也不会超负荷。 另外pipe()会返回目标流的引用，支持链式操做，假设b是个转换流： a.pipe(b).pipe(c)` unpipe()则是用于解绑以前绑定的可写流。 上面的例子能够用data事件改写：

//....
server.on('request', (request, response) => {
    const stream = fs.createReadStream('./file.txt')
    stream.on("data", (chunk) => {
        response.write(chunk)
    })
    stream.on("end", () => {
        response.end() // 使用pipe的话默认会在可读流触发end事件后调用end()结束写入
    })
})
//...
复制代码

不过这样写可能会让可写流超负荷，这就要引入drain的概念了

四、drain和finish事件

流类型：可写流 若是可写流调用write() 返回 false，说明写的太快了，不能再往里面写了。当能够继续写入数据到流时会触发 'drain' 事件。对于上面的例子，咱们来测试一下drain事件是否触发：

//....
server.on('request', (request, response) => {
    const stream = fs.createReadStream('./file.txt')
    stream.on("data", (chunk) => {
        response.write(chunk)
    })
    stream.on("end", () => {
        response.end()
    })
    response.on("drain", () => {
        console.log('能够写了')
  })
})
//...
复制代码

文件file.txt大小100kb左右，运行后看到drain事件触发了4次。虽然写的太快了，可是从http响应的结果看，数据并无丢失。查了一下文档看到有这样的说明：

当流还未被排空时，调用write()会缓冲chunk，并返回false。 一旦全部当前缓冲的数据块都被排空了（被操做系统接收并传输），则触发'drain'事件。 建议一旦write()返回 false，则再也不写入任何数据块，直到'drain'事件被触发。 当流还未被排空时，也是能够调用write()，Node.js 会缓冲全部被写入的数据块，直到达到最大内存占用，这时它会无条件停止。

因此write()返回 false时就不要再往里面写数据了。上面的例子能够这样优化：

server.on('request', (request, response) => {
    conststream = fs.createReadStream('./file.txt')
    let ok = true
    stream.on("data", (chunk) => {
        ok = response.write(chunk)
        if(!ok) {
            stream.pause()
            console.log('别写了')
            ok = true
        }
    })
    stream.on("end", () => {
        response.end()
    })
    response.on("drain", () => {
        console.log('能够写了')
        stream.resume()
    })
})
复制代码

这样写有点麻烦，仍是直接用pipe()比较方便。 这里其实涉及到流缓冲的概念和背压问题，能够查看相关文档进一步学习。 finish事件在调用end() 且缓冲数据都已传给底层系统以后触发。

五、pause()和resume()

流类型：可读流 流动态和静止态的切换，改变data事件是否触发

六、write()和end()

流类型：可写流 上面的例子已有涉及到。write是写入数据到可写流，end代表写入完毕，之久不能再调用write了