深刻理解Node中可讀流和可寫流

時間  2019-11-21
標籤 深刻 理解 node 可讀 简体版
原文   原文鏈接

流是什麼？

這個字進入我腦海我第一時間想到的是一句詩，抽刀斷水水更流，舉杯消愁...額，今天的主角是流。很差意思差點跑題了，嗯，流是一個抽象接口，被 Node 中的不少對象所實現。好比HTTP服務器request和response對象都是流。本人最近研究node，特地記下，分享一下。node

對於流，官方文檔是這樣描述的：緩存

流（stream）在 Node.js 中是處理流數據的抽象接口（abstract interface）bash

Node.js 中有四種基本的流類型：服務器

  • Readable - 可讀的流 (例如 fs.createReadStream()).
  • Writable - 可寫的流 (例如 fs.createWriteStream()).
  • Duplex - 可讀寫的流 (例如 net.Socket).
  • Transform - 在讀寫過程當中能夠修改和變換數據的 Duplex 流 (例如 zlib.createDeflate())

今天主要分享的是node可讀流和可寫流異步

可寫流

先上個流程圖讓你們直觀瞭解整個流程 函數

Writable.png

  • Open（）後write（）開始寫入
  • 判斷是否底層寫入和緩存區是否小於最高水位線同步或異步進行
  • 若是底層在寫入中放到緩存區裏面，不然就調用底層_write()
  • 成功寫入後判斷緩存區是否有數據，若是有在寫入則存加入緩衝區隊列中，緩衝區排空後觸發 drain 事件；
  • 當一個流不處在 drain 的狀態， 對 write() 的調用會緩存數據塊， 而且返回 false。一旦全部當前全部緩存的數據塊都排空了（被操做系統接受來進行輸出）， 那麼 'drain' 事件就會被觸發，一旦 write() 返回 false， 在 'drain' 事件觸發前， 不能寫入任何數據塊。

可寫流的模擬實現：

let EventEmitter = require('events');

class WriteStream extends EventEmitter {
    constructor(path, options) {
        super(path, options);
        this.path = path;
        this.flags = options.flags || 'w';
        this.mode = options.mode || 0o666;
        this.start = options.start || 0;
        this.pos = this.start;//文件的寫入索引
        this.encoding = options.encoding || 'utf8';
        this.autoClose = options.autoClose;
        this.highWaterMark = options.highWaterMark || 16 * 1024;
        this.buffers = [];//緩存區，源碼用的鏈表
        this.writing = false;//表示內部正在寫入數據
        this.length = 0;//表示緩存區字節的長度
        this.open();
    }

    open() {
        fs.open(this.path, this.flags, this.mode, (err, fd) => {
            if (err) {
                if (this.autoClose) {
                    this.destroy();
                }
                return this.emit('error', err);
            }
            this.fd = fd;
            this.emit('open');
        });
    }

    //若是底層已經在寫入數據的話，則必須當前要寫入數據放在緩衝區裏
    write(chunk, encoding, cb) {
        chunk = Buffer.isBuffer(chunk)?chunk:Buffer.from(chunk,this.encoding);
        let len = chunk.length;
        //緩存區的長度加上當前寫入的長度
        this.length += len;
        //判斷當前最新的緩存區是否小於最高水位線
        let ret = this.length < this.highWaterMark;
        if (this.writing) {//表示正在向底層寫數據，則當前數據必須放在緩存區裏
            this.buffers.push({
                chunk,
                encoding,
                cb
            });
        } else {//直接調用底層的寫入方法進行寫入
            //在底層寫完當前數據後要清空緩存區
            this.writing = true;
            this._write(chunk, encoding, () => this.clearBuffer());
        }
        return ret;
    }

    clearBuffer() {
        //取出緩存區中的第一個buffer
        //8 7
        let data = this.buffers.shift();
        if(data){
            this._write(data.chunk,data.encoding,()=>this.clearBuffer())
        }else{
            this.writing = false;
            //緩存區清空了
            this.emit('drain');
        }
    }

    _write(chunk, encoding, cb) {
       if(typeof this.fd != 'number'){
           return this.once('open',()=>this._write(chunk, encoding, cb));
       }
        fs.write(this.fd,chunk,0,chunk.length,this.pos,(err,bytesWritten)=>{
            if(err){
                if(this.autoClose){
                    this.destroy();
                    this.emit('error',err);
                }
            }
            this.pos += bytesWritten;
            //寫入多少字母，緩存區減小多少字節
            this.length -= bytesWritten;
            cb && cb();
       })
    }

    destroy() {
        fs.close(this.fd, () => {
            this.emit('close');
        })
    }
}
module.exports = WriteStream;
複製代碼

可讀流

可讀流事實上工做在下面兩種模式之一：flowing 和 paused

  • 在 flowing 模式下， 可讀流自動從系統底層讀取數據，並經過 EventEmitter 接口的事件儘快將數據提供給應用。
  • 在 paused 模式下，必須顯式調用 stream.read() 方法來從流中讀取數據片斷。

flowing 流動模式

流動模式比較簡單，代碼實現以下：ui

let EventEmitter = require('events');
let fs = require('fs');
class ReadStream extends EventEmitter {
    constructor(path, options) {
        super(path, options);
        this.path = path;
        this.flags = options.flags || 'r';
        this.mode = options.mode || 0o666;
        this.highWaterMark = options.highWaterMark || 64 * 1024;
        this.pos = this.start = options.start || 0;
        this.end = options.end;
        this.encoding = options.encoding;
        this.flowing = null;
        this.buffer = Buffer.alloc(this.highWaterMark);
        this.open();//準備打開文件讀取
        //當給這個實例添加了任意的監聽函數時會觸發newListener
        this.on('newListener',(type,listener)=>{
            //若是監聽了data事件，流會自動切換的流動模式
            if(type == 'data'){
              this.flowing = true;
              this.read();
            }
        });
    }
    read(){
        if(typeof this.fd != 'number'){
            return this.once('open',()=>this.read());
        }
        let howMuchToRead = this.end?Math.min(this.end - this.pos + 1,this.highWaterMark):this.highWaterMark;
        //this.buffer並非緩存區
        console.log('howMuchToRead',howMuchToRead);
        fs.read(this.fd,this.buffer,0,howMuchToRead,this.pos,(err,bytes)=>{//bytes是實際讀到的字節數
            if(err){
                if(this.autoClose)
                    this.destroy();
                return this.emit('error',err);
            }
            if(bytes){
                let data = this.buffer.slice(0,bytes);
                this.pos += bytes;
                data = this.encoding?data.toString(this.encoding):data;
                this.emit('data',data);
                if(this.end && this.pos > this.end){
                   return this.endFn();
                }else{
                    if(this.flowing)
                      this.read();
                }
            }else{
                return this.endFn();
            }

        })
    }
    endFn(){
        this.emit('end');
        this.destroy();
    }
    open() {
        fs.open(this.path,this.flags,this.mode,(err,fd)=>{
           if(err){
               if(this.autoClose){
                   this.destroy();
                   return this.emit('error',err);
               }
           }
           this.fd = fd;
           this.emit('open');
        })
    }
    destroy(){
        fs.close(this.fd,()=>{
            this.emit('close');
        });
    }
    pipe(dest){
        this.on('data',data=>{
            let flag = dest.write(data);
            if(!flag){
                this.pause();
            }
        });
        dest.on('drain',()=>{
            this.resume();
        });
    }
    //可讀流會進入流動模式，當暫停的時候，
    pause(){
        this.flowing = false;
    }
    resume(){
       this.flowing = true;
       this.read();
    }
}
module.exports = ReadStream;
複製代碼

paused 暫停模式：

暫停模式邏輯有點複雜, 畫了一張圖梳理一下this

Readable.png

_read 方法是把數據存在緩存區中，由於是異步 的，流是經過readable事件來通知消耗方的。 說明一下，流中維護了一個緩存，當緩存中的數據足夠多時，調用read()不會引發_read()的調用，即不須要向底層請求數據。state.highWaterMark是給緩存大小設置的一個上限閾值。若是取走n個數據後，緩存中保有的數據不足這個量，便會從底層取一次數據spa

暫停模式代碼模擬實現：操作系統

let fs = require('fs');
let EventEmitter = require('events');

class ReadStream extends EventEmitter {
    constructor(path, options) {
        super(path, options);
        this.path = path;
        this.highWaterMark = options.highWaterMark || 64 * 1024;
        this.buffer = Buffer.alloc(this.highWaterMark);
        this.flags = options.flags || 'r';
        this.encoding = options.encoding;
        this.mode = options.mode || 0o666;
        this.start = options.start || 0;
        this.end = options.end;
        this.pos = this.start;
        this.autoClose = options.autoClose || true;
        this.bytesRead = 0;
        this.closed = false;
        this.flowing;
        this.needReadable = false;
        this.length = 0;
        this.buffers = [];
        this.on('end', function () {
            if (this.autoClose) {
                this.destroy();
            }
        });
        this.on('newListener', (type) => {
            if (type == 'data') {
                this.flowing = true;
                this.read();
            }
            if (type == 'readable') {
                this.read(0);
            }
        });
        this.open();
    }

    open() {
        fs.open(this.path, this.flags, this.mode, (err, fd) => {
            if (err) {
                if (this.autoClose) {
                    this.destroy();
                    return this.emit('error', err);
                }
            }
            this.fd = fd;
            this.emit('open');
        });
    }

    read(n) {
        if (typeof this.fd != 'number') {
            return this.once('open', () => this.read());
        }
        n = parseInt(n, 10);
        if (n != n) {
            n = this.length;
        }
        if (this.length == 0)
            this.needReadable = true;
        let ret;
        if (0 < n < this.length) {
            ret = Buffer.alloc(n);
            let b;
            let index = 0;
            while (null != (b = this.buffers.shift())) {
                for (let i = 0; i < b.length; i++) {
                    ret[index++] = b[i];
                    if (index == ret.length) {
                        this.length -= n;
                        b = b.slice(i + 1);
                        this.buffers.unshift(b);
                        break;
                    }
                }
            }
            if (this.encoding) ret = ret.toString(this.encoding);
        }
        //數據存緩存區中
        let _read = () => {
            let m = this.end ? Math.min(this.end - this.pos + 1, this.highWaterMark) : this.highWaterMark;
            fs.read(this.fd, this.buffer, 0, m, this.pos, (err, bytesRead) => {
                if (err) {
                    return
                }
                let data;
                if (bytesRead > 0) {
                    data = this.buffer.slice(0, bytesRead);
                    this.pos += bytesRead;
                    this.length += bytesRead;
                    if (this.end && this.pos > this.end) {
                        if (this.needReadable) {
                            this.emit('readable');
                        }

                        this.emit('end');
                    } else {
                        this.buffers.push(data);
                        if (this.needReadable) {
                            this.emit('readable');
                            this.needReadable = false;
                        }
                    }
                } else {
                    if (this.needReadable) {
                        this.emit('readable');
                    }
                    return this.emit('end');
                }
            })
        }
        if (this.length == 0 || (this.length < this.highWaterMark)) {
            _read(0);
        }
        return ret;
    }

    destroy() {
        fs.close(this.fd, (err) => {
            this.emit('close');
        });
    }

    pause() {
        this.flowing = false;
    }

    resume() {
        this.flowing = true;
        this.read();
    }

    pipe(dest) {
        this.on('data', (data) => {
            let flag = dest.write(data);
            if (!flag) this.pause();
        });
        dest.on('drain', () => {
            this.resume();
        });
        this.on('end', () => {
            dest.end();
        });
    }
}
module.exports = ReadStream;
複製代碼

小弟我能力有限，歡迎各位大神指點，謝謝~

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程