初探Node中的stream

learn_shifeng 發(fā)布于2019-08-26 11:35 / 2088人閱讀

摘要：則會(huì)立馬再次讀取填滿(mǎn)最高水位線可寫(xiě)流這個(gè)方法向底層系統(tǒng)寫(xiě)入數(shù)據(jù)，并在數(shù)據(jù)處理完畢后調(diào)用所給的回調(diào)。返回值表示你是否應(yīng)該繼續(xù)立即寫(xiě)入。最好的辦法是等待事件后，再寫(xiě)入數(shù)據(jù)。緩存區(qū)未滿(mǎn)寫(xiě)入方法是同步的，但是寫(xiě)入文件的過(guò)程異步的。

Stream流有以下四種類(lèi)型：

Readable - 可讀操作

Writable - 可寫(xiě)操作

Duplex - 可讀可寫(xiě)操作

Transform - 操作被寫(xiě)入數(shù)據(jù)，然后讀出結(jié)果

可讀流（Readable stream）

可讀流（Readable stream）接口是對(duì)你正在讀取的數(shù)據(jù)的來(lái)源的抽象。換句話說(shuō)，數(shù)據(jù)來(lái)來(lái)自可讀流（Readable stream）不會(huì)分發(fā)數(shù)據(jù)，直到你表明準(zhǔn)備就緒。
可讀流（Readable stream）有2種模式: 流動(dòng)模式（flowing mode）和暫停模式（paused mode）. 流動(dòng)模式（flowing mode）時(shí),盡快的從底層系統(tǒng)讀取數(shù)據(jù)并提供給你的程序。暫停模式（paused mode）時(shí), 你必須明確的調(diào)用 stream.read() 來(lái)讀取數(shù)據(jù)。暫停模式（paused mode）是默認(rèn)模式。
可以通過(guò)下面幾個(gè)方法，將流切換到流動(dòng)模式（flowing mode）。

let fs = require("fs");
/**
 * 所有初始工作模式為 paused 的 Readable 流，可以通過(guò)下面三種途徑切換到 flowing 模式：
 監(jiān)聽(tīng) "data" 事件
 調(diào)用 stream.resume() 方法
 調(diào)用 stream.pipe() 方法將數(shù)據(jù)發(fā)送到 Writable
 */
let rs = fs.createReadStream("./1.txt",{
    highWaterMark:3
});
/*
269
 stream.emit("data", chunk);
    stream.read(0);
rs.on("data",function (data) {
    console.log(data);
});
rs.on("end",function () {
    console.log("end");
});*/
//當(dāng)你監(jiān)聽(tīng) readable事件的時(shí)候，會(huì)進(jìn)入暫停模式
//當(dāng)監(jiān)聽(tīng)readable事件的時(shí)候，可讀流會(huì)馬上去向底層讀取文件，然后把讀到文件的文件放在緩存區(qū)里const state = this._readableState;
//self.read(0); 只填充緩存，但是并不會(huì)發(fā)射data事件,但是會(huì)發(fā)射stream.emit("readable");事件
//this._read(state.highWaterMark); 每次調(diào)用底層的方法讀取的時(shí)候是讀取3個(gè)字節(jié)
rs.on("readable",function(){
    //length就是指得緩存區(qū)數(shù)據(jù)的大小
    // state.length +=  chunk.length;==3
    console.log(rs._readableState.length);
    //read如果不加參數(shù)表示讀取整個(gè)緩存區(qū)數(shù)據(jù)
    //讀取一個(gè)字段,如果可讀流發(fā)現(xiàn)你要讀的字節(jié)小于等于緩存字節(jié)大小，則直接返回
    let ch = rs.read(1);
    console.log(ch);
    console.log(rs._readableState.length);
   /* ch = rs.read(1);
    console.log(ch);
    console.log(rs._readableState.length);*/
    //當(dāng)你讀完指定的字節(jié)后，如果可讀流發(fā)現(xiàn)剩下的字節(jié)已經(jīng)比最高水位線小了。則會(huì)立馬再次讀取填滿(mǎn) 最高水位線
    setTimeout(function(){
        console.log(rs._readableState.length);
    },200)
});

可寫(xiě)流（Writable stream ）

這個(gè)方法向底層系統(tǒng)寫(xiě)入數(shù)據(jù)，并在數(shù)據(jù)處理完畢后調(diào)用所給的回調(diào)。返回值表示你是否應(yīng)該繼續(xù)立即寫(xiě)入。如果數(shù)據(jù)要緩存在內(nèi)部，將會(huì)返回false。否則返回 true。返回值僅供參考。即使返回 false，你也可能繼續(xù)寫(xiě)。但是寫(xiě)會(huì)緩存在內(nèi)存里，所以不要做的太過(guò)分。最好的辦法是等待drain 事件后，再寫(xiě)入數(shù)據(jù)。

let fs = require("fs");
let ws = fs.createWriteStream("2.txt",{
    flags:"w",
    mode:0o666,
    start:0,
    highWaterMark:3
});
let count = 9;
function write(){
 let flag = true;//緩存區(qū)未滿(mǎn)
    //寫(xiě)入方法是同步的，但是寫(xiě)入文件的過(guò)程 異步的。在真正寫(xiě)入文件后還會(huì)執(zhí)行我們的回調(diào)函數(shù)
 while(flag && count>0){
     console.log("before",count);
     flag = ws.write((count)+"","utf8",(function (i) {
         return ()=>console.log("after",i);
     })(count));
     count--;
 }
}
write();//987
//監(jiān)聽(tīng)緩存區(qū)清空事件
ws.on("drain",function () {
    console.log("drain");
    write();//654 321
});
ws.on("error",function (err) {
    console.log(err);
});
//如果已經(jīng)不再需要寫(xiě)入了，可以調(diào)用end方法關(guān)閉寫(xiě)入流，一旦調(diào)用end方法之后則不能再寫(xiě)入
ws.end();
//write after end
//
ws.write("x");

雙工流（Duplex streams）

雙工流（Duplex streams）是同時(shí)實(shí)現(xiàn)了?Readable?and?Writable?接口。用法詳見(jiàn)下文

let {Duplex} = require("stream");
let index = 0;
let s = Duplex({
    read(){
        if(index++<3)
          this.push("a"); 
          else 
       this.push(null);   
    },
    write(chunk,encoding,cb){
       console.log(chunk.toString().toUpperCase());
       cb();
    }
});
//process.stdin 標(biāo)準(zhǔn)輸入流
//proces.stdout標(biāo)準(zhǔn)輸出流
process.stdin.pipe(s).pipe(process.stdout);

轉(zhuǎn)換流（Transform streams）

它的輸出是從輸入計(jì)算得來(lái)。它實(shí)現(xiàn)了Readable?和?Writable?接口. 用法詳見(jiàn)下文.

let {Transform}  = require("stream");
//轉(zhuǎn)換流是實(shí)現(xiàn)數(shù)據(jù)轉(zhuǎn)換的
let t = Transform({
    transform(chunk,encoding,cb){
        this.push(chunk.toString().toUpperCase());
        cb();
    }
});
process.stdin.pipe(t).pipe(process.stdout);

let {Transform} = require("stream");
let fs = require("fs");
let rs = fs.createReadStream("./user.json");
//普通流里的放的是Buffer,對(duì)象流里放的對(duì)象
let toJSON = Transform({
    readableObjectMode:true,//就可以向可讀流里放對(duì)象
    transform(chunk,encoding,cb){
        //向可讀流里的緩存區(qū)里放
      this.push(JSON.parse(chunk.toString()));
    }
});
let outJSON = Transform({
    writableObjectMode:true,//就可以向可讀流里放對(duì)象
    transform(chunk,encoding,cb){
      console.log(chunk);
      cb();
    }
});
rs.pipe(toJSON).pipe(outJSON);