一个 stream 的性能问题

原文地址： https://zhuanlan.zhihu.com/p/37601326

之前用 neovim 的 node-client 模块发现，传输大数据时会非常的慢，一个 7M 的文件内容解析需要 30s 才能完成。经常一些尝试，终于找到了问题的原因。事实上做为解码的 msgpack-lite 模块并不慢，7M 的内容解析不会超过 200ms。所以做为数据源的 socket 传输数据慢了？其实也不慢，如果你传输数据会发现传输也不会超过 100ms，示例代码：

const fs = require('fs')
const net = require('net')
const socketPath = '/tmp/test'
try {
  fs.unlinkSync(socketPath)
} catch (e) {}
// 你需要下载一个 sqlite3.c 做为测试文件
const stream = fs.createReadStream('/Users/chemzqm/sqlite3.c', {
  highWaterMark: 1024*1024
})

const server = net.createServer(conn => {
  console.log('client connected, send data')
  stream.pipe(conn)
})

server.on('error', err => {
  throw err
})

server.listen({path:socketPath}, () => {
  console.log('server bound')
})

const client = net.createConnection({
  path:socketPath
})

let l = 0
let ts = Date.now()
client.on('data', chunk => {
  l = l + chunk.length
  server.close()
})
client.on('end', () => {
  console.log(`${Date.now() - ts}`)
})

然鹅，你 socket 的 stream 接到 msgpack 的 stream 就会非常慢了。示例：

const msgpack = require('msgpack-lite')
const fs = require('fs')
const net = require('net')
const socketPath = '/tmp/test'
try {
  fs.unlinkSync(socketPath)
} catch (e) {}

// 使用预先导出的 msgpack 格式 data 数据
const readStream = fs.createReadStream('data.msp', {
  highWaterMark: 1024*1024
});

const server = net.createServer(conn => {
  console.log('client connected, send data')
  readStream.pipe(conn)
})

server.on('error', err => {
  throw err
})

server.listen({path:socketPath}, () => {
  console.log('server bound')
})

const client = net.createConnection({
  path:socketPath,
  highWaterMark: 1024*1024
})

const decodeStream = msgpack.createDecodeStream({
  objectMode: true
})

let ts = Date.now()
client.pipe(decodeStream)
client.on('end', () => {
  console.log('client end')
})
decodeStream.on('data', obj => {
  console.log(Date.now() - ts)
  console.log(obj.length)
})

最终我们发现，socket 只会以 8kb 每次来 emit 数据，而 msgpack stream 对于这种方式的数据反应非常迟钝，然后就导致了解析非常缓慢的结果。起初以为设置 socket 的 highWaterMark 到一个大的数值就能解决这个问题，然而发现这个值在 javascript 里面根本起不了效果，它 emit 的 data 永远最大 8kb。不得已，写了个 Transform stream 做为中介：

import { Transform } from 'stream';

const MIN_SIZE = 8 * 1024;

export default class Buffered extends Transform {
  private chunks: Buffer[] | null;
  constructor() {
    super({
      readableHighWaterMark: 10 * 1024 * 1024,
      writableHighWaterMark: 10 * 1024 * 1024,
    } as any);
    this.chunks = null;
  }

  _transform(chunk: Buffer, encoding: any, callback: any) {
    let { chunks } = this;
    if (chunk.length < MIN_SIZE) {
      if (!chunks) return callback(null, chunk);
      chunks.push(chunk);
      this.chunks = null;
      let buf = Buffer.concat(chunks);
      callback(null, buf);
      return;
    }
    if (!chunks) {
      chunks = this.chunks = [chunk];
    } else {
      chunks.push(chunk);
    }
    setTimeout(() => {
      let { chunks } = this;
      if (chunks) {
        this.chunks = null;
        let buf = Buffer.concat(chunks);
        this.push(buf);
      }
    }, 100);
    callback();
  }

  _flush(callback: any) {
    let { chunks } = this;
    if (chunks) {
      this.chunks = null;
      let buf = Buffer.concat(chunks);
      callback(null, buf);
    } else {
      callback();
    }
  }
}

测试结果：

❯ node index.js
Testing msgpack-lite stream with 2.2M data
msgpack-lite time costed: 4207ms
Testing msgpack5 stream with 2.2M data
msgpack5 time costed: 11189ms
Testing msgpack-lite with buffered stream with 2.2M data
msgpack-lite with buffered time costed: 67ms

对比非常明显。猜测可能原因是 msgpack 的 stream 是工作在 object mode 下面的，而我们的数据源给的都是 buffer，它每次 emit data 都尝试解析 object 所以导致了耗时。测试代码在此：chemzqm/stream-issue