摘要：上游水源通過里中的方法流入水電站中。當在接收數據中出現錯誤時發出。暫?？勺x流，不再發出事件恢復可讀流，繼續發出事件把這個可讀流的輸出傳遞給指定的流，兩個流組成一個管道。

該文章整合了多篇網絡文章（整合之處已設置超鏈接，可點擊直接了解原文），目的僅僅是為了和大伙分享，更加通俗易懂的了解流的各個流程的初始。本人也是node的初學菜鳥，有描述錯誤或誤人子弟的地方多請大神們多多指出。

讀緩沖區（readable buffer）：這里的讀是個形容詞，是指可讀流臨時存放data（只能是字符串或者Buffer，不能是數字）的緩沖區。（讀緩沖區就像一個水電站一樣，感覺這樣描述比較好理解flowing、paused模式）
flowing模式：即流動模式，就像打開水電站的水閘一樣，上游的水和下游完完全全連通直到上游來源的數據耗盡。
paused模式：即暫停模式，就像水電站的水閘在你指定的時候（使用stream.read()）才會打開。不過，當你使用read（）打開水閘的時候是一個超自然現象－－－水電站里的水瞬間被抽干，上游的水還沒來得及填充水電站。然后自動關閉水閘，等待你的下一次“惠顧“read（）。
_read：上游水源通過_read里中的push、unshift方法流入水電站中。

readable：在數據塊可以從流中讀取的時候發出。它對應的處理器沒有參數，可以在處理器里調用read([size])方法讀取數據。
data：有數據可讀時發出。它對應的處理器有一個參數，代表數據。如果你只想快快地讀取一個流的數據，給data關聯一個處理器是最方便的辦法。處理器的參數是Buffer對象，如果你調用了Readable的setEncoding(encoding)方法，處理器的參數就是String對象。
end：當數據被讀完時發出。對應的處理器沒有參數。
close：當底層的資源，如文件，已關閉時發出。不是所有的Readable流都會發出這個事件。對應的處理器沒有參數。
error：當在接收數據中出現錯誤時發出。對應的處理器參數是Error的實例，它的message屬性描述了錯誤原因，stack屬性保存了發生錯誤時的堆棧信息。

read([size])：如果你給read方法傳遞了一個大小作為參數，那它會返回指定數量的數據，如果數據不足，就會返回null。如果你不給read方法傳參，它會返回內部緩沖區里的所有數據，如果沒有數據，會返回null，此時有可能說明遇到了文件末尾。read返回的數據可能是Buffer對象，也可能是String對象。
setEncoding(encoding)：給流設置一個編碼格式，用于解碼讀到的數據。調用此方法后，read([size])方法返回String對象。
pause()：暫停可讀流，不再發出data事件
resume()：恢復可讀流，繼續發出data事件
pipe(destination,[options])：把這個可讀流的輸出傳遞給destination指定的Writable流，兩個流組成一個管道。options是一個JS對象，這個對象有一個布爾類型的end屬性，默認值為true，當end為true時，Readable結束時自動結束Writable。注意，我們可以把一個Readable與若干Writable連在一起，組成多個管道，每一個Writable都能得到同樣的數據。這個方法返回destination，如果destination本身又是Readable流，就可以級聯調用pipe（比如我們在使用gzip壓縮、解壓縮時就會這樣，馬上會講到）。
unpipe([destination])：端口與指定destination的管道。不傳遞destination時，斷開與這個可讀流連在一起的所有管道。

流從默認的暫停模式切換到流動模式可以使用以下幾種方式：

通過添加 data 事件監聽器來啟動數據監聽
調用 resume() 方法啟動數據流
調用 pipe() 方法將數據轉接到另一個 可寫流

從流動模式切換為暫停模式又兩種方法：

在流沒有 pipe() 時，調用 pause() 方法可以將流暫停
pipe() 時，需要移除所有 data 事件的監聽，再調用 unpipe() 方法

data listener
readable listener
read()——如果當前緩沖區為空，或者緩沖區并未超出我們設定的最大值，那么就可以繼續準備數據；如果此時正在準備數據（_read()）或者已經結束讀取（push(null)），那么就放棄準備數據。

這里我要比比兩句哈哈。下面的備注來自的原文大神寫的非常的細膩，還包括了源碼的解讀，不過不太適合初入了解流的同志們，把整個思路理清楚了之后更配喲～

paused模式：

1.在paused模式下則讀取全部緩沖區的長度；若讀取的字節數(n)大于設置的緩沖區最大值，則適當擴大緩沖區的大?。J為16k，最大為8m）；若讀取的長度大于當前緩沖區的大小，設置needReadable屬性并準備數據等待下一次讀取。

2.如果當前緩沖區為空，或者緩沖區并未超出我們設定的最大值，那么就可以繼續準備數據；如果此時正在準備數據（_read()）或者已經結束讀取（push(null)），那么就放棄準備數據。

3.針對這個私有方法_read，文檔上有特殊說明，自定義的Readable實現類需要實現這個方法，在該方法中手動添加數據到Readable對象的讀緩沖區，然后進行Readable的讀取。可以理解為_read函數為讀取數據前的準備工作（準備數據），針對的是流的實現者而言。

flowing模式：

1.對于處在flowing模式下的讀取，每次只讀緩沖區中第一個buffer的長度

2.針對這個私有方法_read，文檔上有特殊說明，自定義的Readable實現類需要實現這個方法，在該方法中手動添加數據到Readable對象的讀緩沖區，然后進行Readable的讀取?？梢岳斫鉃開read函數為讀取數據前的準備工作（準備數據），針對的是流的實現者而言。

paused模式：

//這是一個將存放多條json字符串的txt文件讀取成json的例子
const stream = require("stream");
const fs = require("fs");
const util = require("util");

function JSONLineReader(source) {
    stream.Readable.call(this);
    this._source = source;
    this._foundLineEnd = false;
    this._buffer = "";

    source.on("readable", function() {//監聽source什么時候準備好，那么我們就可以用read()或則readable listener去觸發JSONLineReader的_read方法
        this.read();
        // this.on("readable", function(data) {
        //     console.log("readable");
        // });
    }.bind(this))
}

util.inherits(JSONLineReader, stream.Readable);

JSONLineReader.prototype._read = function(size) {
    var chunk;
    var line;
    var lineIndex;
    var result;
    if (this._buffer.length === 0) {
        chunk = this._source.read();
        this._buffer += chunk; //一次就拿完 只是看什么時候push null
    }
    lineIndex = this._buffer.indexOf("
");
    if (lineIndex !== -1) {
        line = this._buffer.slice(0, lineIndex);
        if (line) {
            result = JSON.parse(line);
            this._buffer = this._buffer.slice(lineIndex + 1);
            this.emit("object", result);util.inspect(result))
            this.push(util.inspect(result));
        } else {
            this._buffer = this._buffer.slice(1);
        }
    }

}

let input = fs.createReadStream(__dirname + "/json-lines.txt", {
    encoding: "utf8"
});

var jsonLineReader = new JSONLineReader(input);

jsonLineReader.on("object", function(obj) {
    console.log("pos:", obj);
})

/*json-lines.txt
{"success":false,"code":501}
{"success":true,"code":202}
{"success":false,"code":503}
{"success":true,"code":204}
{"success":false,"code":505}
{"success":true,"code":206}
{"success":false,"code":507}
{"success":true,"code":208}
{"success":false,"code":509}
*/

flowing模式：

let stream = require("stream");
let util = require("util");
util.inherits(flowingReadableDemo, stream.Readable);

function flowingReadableDemo(opt) {
    stream.Readable.call(this, opt);
    this.quotes = ["yessdasdsa", "noasdasdas", "maybe"];
    this._index = 0;
}
flowingReadableDemo.prototype._read = function() {
    if (this._index >= this.quotes.length) {
        this.push(null);
    } else {
        this.push(this.quotes[this._index]);
        this._index += 1;
    }
};
let r = new flowingReadableDemo();
r.on("data", function(data) {
    console.log("Callback read: " + data.toString());
    // flowing狀態下，我們無需執行read，僅需要設置data事件處理函數或者設定導流目標pipe
});
r.on("end", function(data) {
    console.log("No more answers.");
});