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()
}
}
這樣一來，回呼便會在所有資料都處理完畢後觸發了。
