Node.js中的流十分強(qiáng)大，它對(duì)處理潛在的大文件提供了支持，也抽象了一些場(chǎng)景下的數(shù)據(jù)處理和傳遞。正因?yàn)樗绱撕糜?，所以在?shí)戰(zhàn)中我們常?；谒鼇?lái)編寫一些工具 函數(shù)/庫(kù) ，但往往又由于自己對(duì)流的某些特性的疏忽，導(dǎo)致寫出的 函數(shù)/庫(kù) 在一些情況會(huì)達(dá)不到想要的效果，或者埋下一些隱藏的地雷。本文將會(huì)提供兩個(gè)在編寫基于流的工具時(shí)，私以為有些用的兩個(gè)tips。
    一，警惕EVENTEMITTER內(nèi)存泄露
    在一個(gè)可能被多次調(diào)用的函數(shù)中，如果需要給流添加事件監(jiān)聽(tīng)器來(lái)執(zhí)行某些操作。那么則需要警惕添加監(jiān)聽(tīng)器而導(dǎo)致的內(nèi)存泄露：
    '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函數(shù)會(huì)在通過(guò)監(jiān)聽(tīng)error事件和end事件，來(lái)在流報(bào)錯(cuò)或沒(méi)有數(shù)據(jù)時(shí)，完成這個(gè)Promise。然而在執(zhí)行代碼時(shí)，我們很快就會(huì)在控制臺(tái)中看到報(bào)警信息：(node) warning: possible EventEmitter memory leak detected. 11 error listeners added. Use emitter.setMaxListeners() to increase limit.，因?yàn)槲覀冊(cè)诿看握{(diào)用該函數(shù)時(shí)，都為傳入的流添加了一個(gè)額外的error事件監(jiān)聽(tīng)器和end事件監(jiān)聽(tīng)器。為了避免這種潛在的內(nèi)存泄露，我們要確保每次函數(shù)執(zhí)行完畢后，清除所有此次調(diào)用添加的額外監(jiān)聽(tīng)器，保持函數(shù)無(wú)污染：
    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);
      }
     })
    }
    二，保證工具函數(shù)的回調(diào)在處理完畢數(shù)據(jù)后才被調(diào)用
    工具函數(shù)往往會(huì)對(duì)外提供一個(gè)回調(diào)函數(shù)參數(shù)，待處理完流中的所有數(shù)據(jù)后，帶著指定值觸發(fā)，通常的做法是將回調(diào)函數(shù)的調(diào)用掛在流的end事件中，但如果處理函數(shù)是耗時(shí)的異步操作，回調(diào)函數(shù)則可能在所有數(shù)據(jù)處理完畢前被調(diào)用：
    'use strict';
    const fs = require('fs');
    let stream = fs.createReadStream('/Path/to/a/big/file');
    function processSomeData (stream, callback) {
     stream.on('data', (data) => {
      // 對(duì)數(shù)據(jù)進(jìn)行一些異步耗時(shí)操作
      setTimeout(() => console.log(data), 2000);
     });
     stream.on('end', () => {
      // ...
      callback()
     })
    }
    processSomeData(stream, () => console.log('end'));
    以上的代碼callback回調(diào)可能會(huì)在數(shù)據(jù)并未被全部處理時(shí)就被調(diào)用，因?yàn)榱鞯膃nd事件的觸發(fā)時(shí)機(jī)僅僅是在流中的數(shù)據(jù)被讀完時(shí)。所以我們需要額外地對(duì)數(shù)據(jù)是否已處理完進(jìn)行檢查：
    function processSomeData (stream, callback) {
     let count = 0;
     let finished = 0;
     let isEnd = false;
     stream.on('data', (data) => {
      count++;
      // 對(duì)數(shù)據(jù)進(jìn)行一些異步耗時(shí)操作
      setTimeout(() => {
       console.log(data);
       finished++;
       check();
      }, 2000);
     });
     stream.on('end', () => {
      isEnd = true;
      // ...
      check();
     })
     function check () {
      if (count === finished && isEnd) callback()
     }
    }
    這樣一來(lái)，回調(diào)便會(huì)在所有數(shù)據(jù)都處理完畢后觸發(fā)了。