Node.js 中的可读流

下面的代码启动了一个 HTTP 服务器：

require('http')
  .createServer((req, res) => {
    console.log('headers', req.headers)
    req.on('data', data => console.log('data', data.toString()))
    req.on('end', () => console.log(res.end() && 'end'))
  })
  .listen(3000)

此时如果通过 application/x-www-form-urlencoded 发送 HTTP 请求：

curl -X POST 'localhost:3000' -d 'name=keliq'

服务端会收到：

headers {
  host: 'localhost:3000',
  'user-agent': 'curl/7.64.1',
  accept: '*/*',
  'content-length': '10',
  'content-type': 'application/x-www-form-urlencoded'
}
data name=keliq
end

一切正常，没什么问题。那如果把中间的 req.on('data') 去掉的话会怎样？

require('http')
  .createServer((req, res) => {
    console.log('headers', req.headers)
    // req.on('data', data => console.log('data', data.toString()))
    req.on('end', () => console.log(res.end() && 'end'))
  })
  .listen(3000)

这就涉及到了 Node.js 中的可读流相关的知识了。什么是可读流呢？在 Node.js 中，流是对提供数据的来源的一种抽象，例如：

• 服务器接收到的 HTTP 请求 IncomingMessage
• 使用 fs.createReadStream() 创建的文件流
• 标准输入流 process.stdin
• tcp 的 socket

为什么卡住的问题，要从可读流的两种模式说起。在任何时刻，可读流的状态必定处于一下两种模式之一：

1. 流动模式（flowing）
2. 暂停模式（paused）

而所有的可读流都开始于暂停模式，只能通过以下方式切换到流动模式：

• 添加 data 事件句柄
• 调用 stream.resume()
• 调用 stream.pipe()

由于我们移除了 data 事件句柄，导致可读流一直停留在暂停模式，永远不会触发 end 事件，所以客户端会卡住。解决办法就是借助上面的三种方法让流重新回到流动模式：

require('http')
  .createServer((req, res) => {
    console.log('headers', req.headers)
    // req.on('data', data => console.log('data', data.toString())) // 添加 `data` 事件句柄
    // req.resume() // 调用 `stream.resume()`
    // req.pipe(require('fs').createWriteStream('data.txt')) // 调用 `stream.pipe()`
    req.on('end', () => console.log(res.end() && 'end'))
  })
  .listen(3000)

需要注意的是：如果可读流切换到流动模式，且没有可用的消费者来处理数据，则数据将会丢失。 例如上面调用 readable.resume() 时，没有监听 'data' 事件或 'data' 事件句柄已移除。

其实上面可读流的两种模式，是对发生在可读流中更加复杂的内部状态 _readableState 的抽象，因为在任意时刻，可读流会处于以下三种状态之一：

• readable.readableFlowing === null
• readable.readableFlowing === false
• readable.readableFlowing === true

源码里面有这样一段代码：

ObjectDefineProperties(Readable.prototype, {
  readableFlowing: {
    enumerable: false,
    get: function() {
      return this._readableState.flowing;
    },
    set: function(state) {
      if (this._readableState) {
        this._readableState.flowing = state;
      }
    }
  },
})

所以只要手动设置 readableFlowing 为 true 也可以：

require('http')
  .createServer((req, res) => {
    console.log('headers', req.headers)
    req.readableFlowing = true
    req.on('end', () => console.log(res.end() && 'end'))
  })
  .listen(3000)