nodejs中stream(流)學習分享

爲何須要stream

作爲一個前端猿，一直對後端天天能夠任性的操做服務器文件感到羨慕。最近學習node的fs模塊玩的也是意猶未盡好不知足。fs.readFile方法在讀取大文件時沒法勝任，因此流出現了。

stream(流)

個人理解 咱們在網上下載資源時，稍微大一點就要下載個把小時。咱們的帶寬遠遠不能知足從服務端把資源整個端走，因此要把資源拆分紅小塊，一塊一塊的運輸。資源就像水同樣流到了咱們本地。

While it is important for all Node.js users to understand how streams work.(官網說流很重要！)

流的類型

nodejs有四種基本的流類型：

• 1 Readable 可讀流
• 2 Writable 可寫流
• 3 Duplex 可讀可寫流
• 4 Transform 在讀寫過程當中能夠修改和變換數據的Duplex流

1.Readable

可讀流的功能是做爲上游，提供數據給下游。

建立與使用

let {Readable} = require('stream');
let readable = Readable();
/*
Readable({read(){})
*/
let source = ['a','b','c'];
readable.setEncoding('utf8');
readable._read = function () {
    let data = source.shift()||null;
    console.log('read:',data);
    this.push(data);
}
readable.on('end', function () {
  console.log('end')
})
readable.on('data', function (data) {
    console.log(data)
})
/*
輸出：
read: a
read: b
a
read: c
b
read: null
c
end
*/
複製代碼

Readable經過實例_read或read方法,在須要數據時，_read()方法會自動調用。

• _read方法結束的標誌是this.push(null)。
• 當readable綁定data時_read()方法自動調用

data事件：當讀入數據時觸發data事件傳入回調讀到內容。
end事件：「消耗完」，須要知足兩個條件：

• 已經調用push(null)，聲明不會再有任何新的數據產生
• 緩存中的數據也被讀取完

兩種模式(flowing 和 paused)

流動(flowing)模式

flowing 模式下， 可讀流自動從系統底層讀取數據，並經過 EventEmitter 接口的事件儘快將數據提供給應用。
如下條件都可以使readable進入flowing模式：

• 調用resume方法
• 若是以前未調用pause方法進入paused模式，則監聽data事件也會調用resume方法。
• readable.pipe(writable)。pipe中會監聽data事件。

暫停(paused)模式

paused 模式下，必須顯式調用 stream.read() 方法來從流中讀取數據片斷。
可讀流能夠經過下面途徑切換到 paused 模式：

• 若是不存在管道目標（pipe destination），能夠經過調用 stream.pause() 方法實現。
• 若是存在管道目標，能夠經過取消 'data' 事件監聽，並調用 stream.unpipe() 方法移除全部管道目標來實現。

---

Readable流程

1. 經過read或_read方法讀入數據
2. 將數據push到當前對象
3. 觸發emit data方法

   

深刻理解readable
衆所周知node i/o操做一般採用異步操做。在讀取數據時會把數據寫入緩存區。

read方法

在建立流時，會設置一個highWaterMark參數建立最高水位線。

###自定義可讀流

var stream = require('stream');
var util = require('util');
util.inherits(Counter, stream.Readable);
function Counter(options) {
    stream.Readable.call(this, options);
    this._index = 0;
}
Counter.prototype._read = function() {
    if(this._index++<3){
        this.push(this._index+'');
    }else{
        this.push(null);
    }
};
var counter = new Counter();

counter.on('data', function(data){
    console.log("讀到數據: " + data.toString());//no maybe
});
counter.on('end', function(data){
    console.log("讀完了");
});
複製代碼

2.Writable

可寫流是對數據寫入'目的地'的一種抽象。可寫流的功能是做爲下游，消耗上游提供的數據。

let {Writable} = require('stream');
var writable = Writable({
    write: function (data,_,next) {
        console.log(data);
        next&&next();
    }
})
/*
Writable._write(data,_,next)
*/
writable.write('a');
writable.write('b');
writable.write('c');
writable.end();
/*
<Buffer 61>
<Buffer 62>
<Buffer 63>
*/
複製代碼

write方法

• write()或_write()的第三個參數next爲回調函數，調用next()表示寫入完成，開始寫下一個數據。
• 必須調用end()方法來告訴writable，全部數據均已寫入。

自定義可寫流

var stream = require('stream');
var util = require('util');
util.inherits(Writer, stream.Writable);
let stock = [];
function Writer(opt) {
    stream.Writable.call(this, opt);
}
Writer.prototype._write = function(chunk, encoding, callback) {
    setTimeout(()=>{
        stock.push(chunk.toString('utf8'));
        console.log("增長: " + chunk);
        callback();
    },500)
};
var w = new Writer();
for (var i=1; i<=5; i++){
    w.write("項目:" + i, 'utf8');
}
w.end("結束寫入",function(){
    console.log(stock);
});
複製代碼

3.管道流

從readable讀出數據，writeable寫入數據

建立與使用

const stream = require('stream')

var index = 0;
const readable = stream.Readable({
    highWaterMark: 2,
    read: function () {
        process.nextTick(() => {
            console.log('push', ++index)
            this.push(index+'');
        })
    }
})

const writable = stream.Writable({
    highWaterMark: 2,
    write: function (chunk, encoding, next) {
        console.log('寫入:', chunk.toString())
    }
})

readable.pipe(writable);
複製代碼

4.Duplex

const {Duplex} = require('stream');
const inoutStream = new Duplex({
    write(chunk, encoding, callback) {
        console.log(chunk.toString());
        callback();
    },
    read(size) {
        this.push((++this.index)+'');
        if (this.index > 3) {
            this.push(null);
        }
    }
});

inoutStream.index = 0;
process.stdin.pipe(inoutStream).pipe(process.stdout);
複製代碼

5.Transform

const {Transform} = require('stream');

const upperCase = new Transform({
    transform(chunk, encoding, callback) {
        this.push(chunk.toString().toUpperCase());
        callback();
    }
});

process.stdin.pipe(upperCase).pipe(process.stdout);
/*
實現大小寫轉換
inpt: a
A
*/
複製代碼

---

最近在學習nodejs，通過整理寫下學習筆記。歡迎同窗溝通學習心得，若有錯誤的地方也歡迎指正。