Node.js中的流非常壮大，它对处置惩罚潜伏的大文件供应了支撑，也笼统了一些场景下的数据处置惩罚和通报。正由于它云云好用，所以在实战中我们经常基于它来编写一些东西 函数/库 ，但每每又由于本身对流的某些特征的忽视，致使写出的 函数/库 在一些状况会达不到想要的结果，或许埋下一些隐蔽的地雷。本文将会供应两个在编写基于流的东西时，私认为有些用的两个tips。

一，小心EventEmitter内存泄漏

在一个能够被屡次挪用的函数中，假如须要给流增加事宜监听器来实行某些操纵。那末则须要小心增加监听器而致使的内存泄漏：

'use strict';
const fs = require('fs');
const co = require('co');

function getSomeDataFromStream (stream) {
  let data = stream.read();
  if (data) return Promise.resolve(data);

  if (!stream.readable) return Promise.resolve(null);

  return new Promise((resolve, reject) => {
    stream.once('readable', () => resolve(stream.read()));
    stream.on('error', reject);
    stream.on('end', resolve);
  })
}

let stream = fs.createReadStream('/Path/to/a/big/file');

co(function *() {
  let chunk;
  while ((chunk = yield getSomeDataFromStream(stream)) !== null) {
    console.log(chunk);
  }
}).catch(console.error);

在上述代码中，getSomeDataFromStream函数会在经由过程监听error事宜和end事宜，来在流报错或没有数据时，完成这个Promise。然而在实行代码时，我们很快就会在掌握台中看到报警信息：(node) warning: possible EventEmitter memory leak detected. 11 error listeners added. Use emitter.setMaxListeners() to increase limit.，由于我们在每次挪用该函数时，都为传入的流增加了一个分外的error事宜监听器和end事宜监听器。为了防止这类潜伏的内存泄漏，我们要确保每次函数实行终了后，消灭一切此次挪用增加的分外监听器，坚持函数无污染：

function getSomeDataFromStream (stream) {
  let data = stream.read();
  if (data) return Promise.resolve(data);

  if (!stream.readable) return Promise.resolve(null);

  return new Promise((resolve, reject) => {
    stream.once('readable', onData);
    stream.on('error', onError);
    stream.on('end', done);

    function onData () {
      done();
      resolve(stream.read());
    }

    function onError (err) {
      done();
      reject(err);
    }

    function done () {
      stream.removeListener('readable', onData);
      stream.removeListener('error', onError);
      stream.removeListener('end', done);
    }
  })
}

二，保证东西函数的回调在处置惩罚终了数据后才被挪用

东西函数每每会对外供应一个回调函数参数，待处置惩罚完流中的一切数据后，带着指定值触发，一般的做法是将回调函数的挪用挂在流的end事宜中，但假如处置惩罚函数是耗时的异步操纵，回调函数则能够在一切数据处置惩罚终了前被挪用：

'use strict';
const fs = require('fs');

let stream = fs.createReadStream('/Path/to/a/big/file');

function processSomeData (stream, callback) {
  stream.on('data', (data) => {
    // 对数据举行一些异步耗时操纵
    setTimeout(() => console.log(data), 2000);
  });

  stream.on('end', () => {
    // ...
    callback()
  })
}

processSomeData(stream, () => console.log('end'));

以上的代码callback回调能够会在数据并未被悉数处置惩罚时就被挪用，由于流的end事宜的触发机遇仅仅是在流中的数据被读完时。所以我们须要分外埠对数据是不是已处置惩罚完举行检查：

function processSomeData (stream, callback) {
  let count = 0;
  let finished = 0;
  let isEnd = false;

  stream.on('data', (data) => {
    count++;
    // 对数据举行一些异步耗时操纵
    setTimeout(() => {
      console.log(data);
      finished++;
      check();
    }, 2000);
  });

  stream.on('end', () => {
    isEnd = true;
    // ...
    check();
  })

  function check () {
    if (count === finished && isEnd) callback()
  }
}

这样一来，回调便会在一切数据都处置惩罚终了后触发了。