碼上快樂

相關內容    簡體    繁體

nodejs-stream部分

本文轉載自   查看原文   2018-10-17 17:03   1149    node文檔學習

參考：

https://blog.csdn.net/eeewwwddd/article/details/81042225

http://nodejs.cn/api/stream.html#stream_writable_write_chunk_encoding_callback

 

流（stream）是 Node.js 中處理流式數據的抽象接口。 stream 模塊提供了一些 API，用於構建實現了流接口的對象。

Node.js 提供了多種流對象。 例如，HTTP 服務器的請求和 process.stdout 都是流的實例。

流可以是可讀的、可寫的、或者可讀可寫的。 所有的流都是 EventEmitter 的實例，即可以通過事件的監聽得以觸發事件並執行一定的操作，如：

  req.on('data', (chunk) => {
    body += chunk;
  });

stream 模塊可以通過以下方式使用：

const stream = require('stream');

盡管理解流的工作方式很重要，但是 stream 模塊本身主要用於開發者創建新類型的流實例。

對於以消費流對象為主的開發者，極少需要直接使用 stream 模塊。

 

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

    Writable - 可寫入數據的流（例如 fs.createWriteStream()）。
    Readable - 可讀取數據的流（例如 fs.createReadStream()）。
    Duplex - 可讀又可寫的流（例如 net.Socket）。
    Transform - 在讀寫過程中可以修改或轉換數據的 Duplex 流（例如 zlib.createDeflate()）

 

兩種模式

 二進制模式

每個分塊都是buffer、string對象

對象模式

Node.js 創建的流都是運作在字符串和 Buffer（或 Uint8Array）上。 當然，流的實現也可以使用其它類型的 JavaScript 值（除了 null）。 這些流會以“對象模式”進行操作。

當創建流時，可以使用 objectMode 選項把流實例切換到對象模式。 將已存在的流切換到對象模式是不安全的。

 

 比如如果想創建一個的可以壓入任意形式數據的可讀流，只要在創建流的時候設置參數objectMode為true即可，例如：Readable({ objectMode: true })

如果readable stream寫入的是字符串，那么字符串會默認轉換為Buffer，如果在創建流的時候設置Writable({ decodeStrings: false })參數，那么不會做轉換。

如果readable stream寫入的數據是對象，那么需要這樣創建writable stream，Writable({ objectMode: true })

⚠️就是如果輸入的數據並不是Buffer（或 Uint8Array）格式的時候，那么在創建這個流的時候就要將其設置為對象模式，即設置其的objectMode: true，舉例：

const DuplexStream = require('readable-stream').Duplex
const inherits = require('util').inherits

module.exports = PostMessageStream

inherits(PostMessageStream, DuplexStream)

function PostMessageStream (opts) {
  DuplexStream.call(this, {
  objectMode: true,
  })
...
}

 

緩沖

可寫流和可讀流都會在內部的緩沖器中存儲數據，可以分別使用的 writable.writableBuffer 或 readable.readableBuffer 來獲取。

可緩沖的數據大小取決於傳入流構造函數的 highWaterMark 選項。 對於普通的流，highWaterMark 指定了字節的總數。 對於對象模式的流，highWaterMark 指定了對象的總數。

當調用 stream.push(chunk) 時，數據會被緩沖在可讀流中。 如果流的消費者沒有調用 stream.read()，則數據會保留在內部隊列中直到被消費。

一旦內部的可讀緩沖的總大小達到 highWaterMark 指定的閾值時，流會暫時停止從底層資源讀取數據，直到當前緩沖的數據被消費 （也就是說，流會停止調用內部的用於填充可讀緩沖的 readable._read()）。

當調用 writable.write(chunk) 時，數據會被緩沖在可寫流中。 當內部的可寫緩沖的總大小小於 highWaterMark 設置的閾值時，調用 writable.write() 會返回 true。 一旦內部緩沖的大小達到或超過 highWaterMark 時，則會返回 false。

stream API 的主要目標，特別是 stream.pipe()，是為了限制數據的緩沖到可接受的程度，也就是讀寫速度不一致的源頭與目的地不會壓垮內存。

因為 Duplex 和 Transform 都是可讀又可寫的，所以它們各自維護着兩個相互獨立的內部緩沖器用於讀取和寫入， 這使得它們在維護數據流時，讀取和寫入兩邊可以各自獨立地運作。 例如，net.Socket 實例是 Duplex 流，它的可讀端可以消費從 socket 接收的數據，而可寫端則可以將數據寫入到 socket。 因為數據寫入到 socket 的速度可能比接收數據的速度快或者慢，所以在讀寫兩端獨立地進行操作（或緩沖）就顯得很重要了。

 

【1】用於消費流的 API（即讀取流中數據）

test.js

const http = require('http');

const server = http.createServer((req, res) => {
  // req 是一個 http.IncomingMessage 實例，它是可讀流。
  // res 是一個 http.ServerResponse 實例，它是可寫流。

  let body = '';
  // 接收數據為 utf8 字符串，
  // 如果沒有設置字符編碼，則會接收到 Buffer 對象。
  req.setEncoding('utf8');

  // 如果添加了監聽器，則可讀流會觸發 'data' 事件。
  req.on('data', (chunk) => {
    body += chunk;
  });

  // 'end' 事件表明整個請求體已被接收。 
  req.on('end', () => {
    try {
      const data = JSON.parse(body);
      // 響應信息給用戶。
      res.write(typeof data);
      res.end();//end()表示寫結束
    } catch (er) {
      // json 解析失敗。
      res.statusCode = 400;
      return res.end(`錯誤: ${er.message}`);
    }
  });
});

server.listen(1337);

然后在終端使用node test.js運行該服務器

然后在另一個終端使用curl localhost:1337 -d "{}" 連接服務器localhost:1337 ,-d即post數據data為{} ，返回object

curl localhost:1337 -d "{}"       返回object
curl localhost:1337 -d "\"foo\""  返回string
curl localhost:1337 -d "not json" 返回 錯誤: Unexpected token o in JSON at position 1

 

可寫流（比如例子中的 res）會暴露了一些方法，比如 write() 和 end() 用於寫入數據到流。

當數據可以從流讀取時，可讀流會使用 EventEmitter API 來通知應用程序。 從流讀取數據的方式有很多種。

可寫流和可讀流都通過多種方式使用 EventEmitter API 來通訊流的當前狀態。

Duplex 流和 Transform 流都是可寫又可讀的。

對於只需寫入數據到流或從流消費數據的應用程序，並不需要直接實現流的接口，通常也不需要調用 require('stream')

 

《1》可寫流

可寫流是對數據要被寫入的目的地的一種抽象。

可寫流的例子包括：

上面的一些例子事實上是實現了可寫流接口的 Duplex 流。

所有可寫流都實現了 stream.Writable 類定義的接口。

盡管可寫流的具體實例可能略有差別，但所有的可寫流都遵循同一基本的使用模式，如以下例子所示：

const myStream = getWritableStreamSomehow();
myStream.write('一些數據');
myStream.write('更多數據');
myStream.end('完成寫入數據');//說明完成寫入

 

stream.Writable 類

下面介紹幾類事件：

'close' 事件

當流或其底層資源（比如文件描述符）被關閉時觸發。 表明不會再觸發其他事件，也不會再發生操作。

不是所有可寫流都會觸發 'close' 事件。

'drain' 事件

如果調用 stream.write(chunk) 返回 false，可能緩沖區已滿，需要等待，則當有空間可以繼續寫入數據到流時會觸發 'drain' 事件。

// 向可寫流中寫入數據一百萬次。
// 留意背壓（back-pressure）。
function writeOneMillionTimes(writer, data, encoding, callback) {
  let i = 1000000;
  write();
  function write() {
    let ok = true;
    do {
      i--;
      if (i === 0) {
        // 最后一次寫入。
        writer.write(data, encoding, callback);
      } else {
        // 檢查是否可以繼續寫入。 
        // 不要傳入回調，因為寫入還沒有結束。
        ok = writer.write(data, encoding);
      }
    } while (i > 0 && ok);
    if (i > 0) {
      // 被提前中止。
      // 當觸發 'drain' 事件時繼續寫入，繼續運行write()函數。
      writer.once('drain', write);
    }
  }
}
'error' 事件

當寫入數據發生錯誤時觸發。

當觸發 'error' 事件時，流還未被關閉

'finish' 事件

調用 stream.end() 且緩沖數據都已傳給底層系統之后觸發。

const http = require('http');

const server = http.createServer((req, res) => {
  // req 是一個 http.IncomingMessage 實例，它是可讀流。
  // res 是一個 http.ServerResponse 實例，它是可寫流。

  let body = '';
  // 接收數據為 utf8 字符串，
  // 如果沒有設置字符編碼，則會接收到 Buffer 對象。
  req.setEncoding('utf8');

  // 如果添加了監聽器，則可讀流會觸發 'data' 事件。
  req.on('data', (chunk) => {
    body += chunk;
  });

  // 'end' 事件表明整個請求體已被接收。 
  req.on('end', () => {
    try {
      const data = JSON.parse(body);
      // 響應信息給用戶。
      res.write(typeof data);
      res.end();//會觸發finish事件
 res.on('finish', () => {
          console.error('寫入已完成'); });
    } catch (er) {
      // json 解析失敗。
      res.statusCode = 400;
      return res.end(`錯誤: ${er.message}`);
    }
  });
});

server.listen(1337);

運行結果：

'pipe' 事件

當在可讀流上調用 stream.pipe() 時觸發。

var assert = require('assert');
const writer = process.stdout;
const reader = process.stdin;
writer.on('pipe', (src) => {
  console.error('有數據正通過管道流入寫入器');
  assert.equal(src,reader);//兩者相等
  console.log(src);
});
reader.pipe(writer);

返回：

有數據正通過管道流入寫入器
ReadStream {
  connecting: false,
  _hadError: false,
  _handle:
   TTY { owner: [Circular], onread: [Function: onread], reading: false },
  _parent: null,
  _host: null,
  _readableState:
   ReadableState {
     objectMode: false,//非對象模式
     highWaterMark: 0,
     buffer: BufferList { length: 0 },
     length: 0,
     pipes:
      WriteStream {
        connecting: false,
        _hadError: false,
        _handle: [TTY],
        _parent: null,
        _host: null,
        _readableState: [ReadableState],
        readable: false,
        _events: [Object],
        _eventsCount: 7,
        _maxListeners: undefined,
        _writableState: [WritableState],
        writable: true,
        allowHalfOpen: false,
        _sockname: null,
        _writev: null,
        _pendingData: null,
        _pendingEncoding: '',
        server: null,
        _server: null,
        columns: 80,
        rows: 24,
        _type: 'tty',
        fd: 1,
        _isStdio: true,
        destroySoon: [Function: destroy],
        _destroy: [Function],
        [Symbol(asyncId)]: 2,
        [Symbol(lastWriteQueueSize)]: 0,
        [Symbol(timeout)]: null,
        [Symbol(kBytesRead)]: 0,
        [Symbol(kBytesWritten)]: 0 },
     pipesCount: 1,
     flowing: true,
     ended: false,
     endEmitted: false,
     reading: false,
     sync: false,
     needReadable: true,
     emittedReadable: false,
     readableListening: false,
     resumeScheduled: true,
     emitClose: false,
     destroyed: false,
     defaultEncoding: 'utf8',
     awaitDrain: 0,
     readingMore: false,
     decoder: null,
     encoding: null },
  readable: true,
  _events:
   { end: [ [Function: onReadableStreamEnd], [Function] ],
     pause: [Function],
     data: [Function: ondata] },
  _eventsCount: 3,
  _maxListeners: undefined,
  _writableState:
   WritableState {
     objectMode: false,
     highWaterMark: 0,
     finalCalled: false,
     needDrain: false,
     ending: false,
     ended: false,
     finished: false,
     destroyed: false,
     decodeStrings: false,
     defaultEncoding: 'utf8',
     length: 0,
     writing: false,
     corked: 0,
     sync: true,
     bufferProcessing: false,
     onwrite: [Function: bound onwrite],
     writecb: null,
     writelen: 0,
     bufferedRequest: null,
     lastBufferedRequest: null,
     pendingcb: 0,
     prefinished: false,
     errorEmitted: false,
     emitClose: false,
     bufferedRequestCount: 0,
     corkedRequestsFree:
      { next: null,
        entry: null,
        finish: [Function: bound onCorkedFinish] } },
  writable: false,
  allowHalfOpen: false,
  _sockname: null,
  _writev: null,
  _pendingData: null,
  _pendingEncoding: '',
  server: null,
  _server: null,
  isRaw: false,
  isTTY: true,
  fd: 0,
  [Symbol(asyncId)]: 5,
  [Symbol(lastWriteQueueSize)]: 0,
  [Symbol(timeout)]: null,
  [Symbol(kBytesRead)]: 0,
  [Symbol(kBytesWritten)]: 0 }
View Code

 

'unpipe' 事件

當在可讀流上調用 stream.unpipe() 時觸發。

當可讀流通過管道流向可寫流發生錯誤時，也會觸發 'unpipe' 事件。

var assert = require('assert');
const writer = process.stdout;
const reader = process.stdin;
writer.on('pipe', (src) => {
  console.error('有數據正通過管道流入寫入器');
  assert.equal(src,reader);
  // console.log(src);
});
writer.on('unpipe', (src) => {
  console.error('已移除可寫流管道');
  assert.equal(src, reader);
});
reader.pipe(writer);//觸發'pipe'事件
reader.unpipe(writer);//觸發'unpipe'事件

返回：

userdeMacBook-Pro:stream-learning user$ node test.js
有數據正通過管道流入寫入器
已移除可寫流管道

 

 

下面是可使用的方法：

writable.write(chunk[, encoding][, callback])
  • chunk <string> | <Buffer> | <Uint8Array> | <any> 要寫入的數據。  對於非對象模式的流chunk 必須是字符串、Buffer 或 Uint8Array。 對於對象模式的流，chunk 可以是任何 JavaScript 值，除了 null。
  • encoding <string> 如果 chunk 是字符串，則指定字符編碼。
  • callback <Function> 當數據塊被輸出到目標后的回調函數。
  • 返回: <boolean> 如果流需要等待 'drain' 事件觸發才能繼續寫入更多數據，則返回 false，否則返回 true。
//wfd文件描述符，一般通過fs.open中獲取
//buffer，要取數據的緩存源
//0，從buffer的0位置開始取
//BUFFER_SIZE，每次取BUFFER_SIZE這么長的長度
//index，每次寫入文件的index的位置
//bytesRead，真實寫入的個數
fs.write(wfd,buffer,0,bytesRead,index,function(err,bytesWrite){

})

writable.write() 寫入數據到流，並在數據被完全處理之后調用 callback。 如果發生錯誤，則 callback 可能被調用也可能不被調用。 為了可靠地檢測錯誤，可以為 'error' 事件添加監聽器。

在接收了 chunk 后，如果內部的緩沖小於創建流時配置的 highWaterMark，則返回 true 。 如果返回 false ，則應該停止向流寫入數據，直到 'drain' 事件被觸發。

當流還未被排空時，調用 write() 會緩沖 chunk，並返回 false。 一旦所有當前緩沖的數據塊都被排空了（被操作系統接收並傳輸），則觸發 'drain' 事件。 建議一旦 write() 返回 false，則不再寫入任何數據塊，直到 'drain' 事件被觸發。 當流還未被排空時，也是可以調用 write()，Node.js 會緩沖所有被寫入的數據塊，直到達到最大內存占用，這時它會無條件中止。 甚至在它中止之前， 高內存占用將會導致垃圾回收器的性能變差和 RSS 變高（即使內存不再需要，通常也不會被釋放回系統）。 如果遠程的另一端沒有讀取數據，TCP 的 socket 可能永遠也不會排空，所以寫入到一個不會排空的 socket 可能會導致遠程可利用的漏洞。 

對於 Transform, 寫入數據到一個不會排空的流尤其成問題，因為 Transform 流默認會被暫停，直到它們被 pipe 或者添加了 'data' 或 'readable' 事件句柄。 

如果要被寫入的數據可以根據需要生成或者取得，建議將邏輯封裝為一個可讀流並且使用 stream.pipe()。 如果要優先調用 write()，則可以使用 'drain' 事件來防止背壓與避免內存問題:

var assert = require('assert'); const writer = process.stdout; // const reader = process.stdin;  function write(data, cb) { if (!writer.write(data)) { writer.once('drain', cb); } else { process.nextTick(cb); } } // 在回調函數被執行后再進行其他的寫入。 write('hello', () => { console.log('完成寫入，可以進行更多的寫入'); });

返回：

node test.js
hello完成寫入，可以進行更多的寫入

 

舉一個例子說明write和drain：

參考https://blog.csdn.net/eeewwwddd/article/details/81042225

  • 如果文件不存在會創建，如果有內容會被清空
  • 讀取到highWaterMark的時候就會輸出
  • 第一次是真的寫到文件 后面就是寫入緩存區 再從緩存區里面去取
path:寫入的文件的路徑
option: 
highWaterMark:水位線，一次可寫入緩存中的字節，一般默認是64k
flags:標識，寫入文件要做的操作，默認是w
encoding:編碼，默認為buffer
start:開始寫入的索引位置
end:結束寫入的索引位置(包括結束位置)
autoClose:寫入完畢是否關閉，默認為true

 

let fs = require('fs')
let ws = fs.createWriteStream('./foo1.txt',{
  flags: 'w',
  encoding: 'utf8',
  start: 0,
  //write的highWaterMark只是用來觸發是不是干了
  highWaterMark: 19 //寫是默認16k，當這里設置的長度小於或者等於我一下子要寫入的字符串長度時，會觸發一次drain，也僅觸發一次，然后將剩余部分的所有內容放入緩存，后面將不會再觸發drain了
})
//返回boolean 每當write一次都會在ws中吃下一個饅頭 當吃下的饅頭數量達到highWaterMark時 就會返回false 吃不下了會把其余放入緩存 其余狀態返回true
//write只能放string或者buffer
var flag = ws.write('today is a good day','utf8',()=>{
  console.log('write');
});
ws.on('drain', ()=>{
    console.log('drain');
});

返回：

node test.js
drain
write

如果改為highWaterMark: 20，大於輸入內容，則不會觸發drain

則返回：

 

node test.js
write

 

 

 

writable.end([chunk][, encoding][, callback])
  • chunk <string> | <Buffer> | <Uint8Array> | <any> 要寫入的數據。 對於非對象模式的流chunk 必須是字符串、Buffer、或 Uint8Array。 對於對象模式的流， chunk 可以是任何 JavaScript 值，除了 null。
  • encoding <string> 如果 chunk 是字符串，則指定字符編碼。
  • callback <Function> 當流結束時的回調函數。
  • 返回: <this>

調用 writable.end() 表明已沒有數據要被寫入可寫流。 可選的 chunk 和 encoding 參數可以在關閉流之前再寫入一塊數據。 如果傳入了 callback 函數，則會做為監聽器添加到 'finish' 事件。

調用 stream.end() 之后再調用 stream.write() 會導致錯誤

writable.cork()

強制把所有寫入的數據都緩沖到內存中。 當調用 stream.uncork() 或 stream.end() 時，緩沖的數據才會被輸出。

當寫入大量小塊數據到流時，內部緩沖可能失效，從而導致性能下降，writable.cork() 主要用於避免這種情況。 對於這種情況，實現了 writable._writev() 的流可以用更優的方式對寫入的數據進行緩沖。

writable.uncork()

將調用 stream.cork() 后緩沖的所有數據輸出到目標。

當使用 writable.cork() 和 writable.uncork() 來管理流的寫入緩沖時，建議使用 process.nextTick() 來延遲調用 writable.uncork()。 通過這種方式，可以對單個 Node.js 事件循環中調用的所有 writable.write() 進行批處理。

 

擴展： process.nextTick()

process.nextTick(callback[, ...args])

  • callback <Function>
  • ...args <any> 調用 callback時傳遞給它的額外參數

process.nextTick()方法將 callback 添加到"next tick 隊列"。 一旦當前事件輪詢隊列的任務全部完成，在next tick隊列中的所有callbacks會被依次調用。

這種方式不是setTimeout(fn, 0)的別名。它更加有效率。事件輪詢隨后的ticks 調用，會在任何I/O事件（包括定時器）之前運行。

舉例：

console.log('start');
process.nextTick(() => {
  console.log('nextTick callback');
});
console.log('scheduled');
// Output:
// start
// scheduled
// nextTick callback

 

回到writable.uncork()，舉例：

var assert = require('assert');
const writer = process.stdout;

writer.cork();
writer.write('一些 ');
writer.write('數據 ');
process.nextTick(() => writer.uncork());如果沒有這一句，運行時沒有輸出結果的

返回：

node test.js
一些 數據

如果一個流上多次調用 writable.cork()，則必須調用同樣次數的 writable.uncork() 才能輸出緩沖的數據。

var assert = require('assert');
const writer = process.stdout;

writer.cork();
writer.write('一些 ');
writer.cork();
writer.write('數據 ');
process.nextTick(() => {
  writer.uncork();
  // 數據不會被輸出，直到第二次調用 uncork()。
  writer.uncork();//注釋掉這一句就不會有輸出，正確輸出為一些 數據 
});
writable.destroy([error])

銷毀流，並觸發 'error' 事件且傳入 error 參數。 調用該方法后，可寫流就結束了，之后再調用 write() 或 end() 都會導致 ERR_STREAM_DESTROYED 錯誤。 實現流時不應該重寫這個方法，而是重寫 writable._destroy()

 

writable.setDefaultEncoding(encoding)

為可寫流設置默認的 encoding。

 

轉自https://blog.csdn.net/eeewwwddd/article/details/81042225

let fs = require('fs')
let EventEmitter = require('events')
//只有第一次write的時候直接用_write寫入文件 其余都是放到cache中 但是len超過了highWaterMark就會返回false告知需要drain 很占緩存
//從第一次的_write開始 回去一直通過clearBuffer遞歸_write寫入文件 如果cache中沒有了要寫入的東西 會根據needDrain來判斷是否觸發干點
class WriteStream extends EventEmitter{
  constructor(path,options = {}){
    super()
    this.path = path
    this.highWaterMark = options.highWaterMark || 64*1024
    this.flags = options.flags || 'r'
    this.start = options.start || 0
    this.pos = this.start
    this.autoClose = options.autoClose || true
    this.mode = options.mode || 0o666
    //默認null就是buffer
    this.encoding = options.encoding || null

    //打開這個文件
    this.open()

    //寫文件的時候需要哪些參數
    //第一次寫入的時候 是給highWaterMark個饅頭 他會硬着頭皮寫到文件中 之后才會把多余吃不下的放到緩存中
    this.writing = false
    //緩存數組
    this.cache = []
    this.callbackList = []
    //數組長度
    this.len = 0
    //是否觸發drain事件
    this.needDrain = false
  }

  clearBuffer(){
    //取緩存中最上面的一個
    let buffer = this.cache.shift()
    if(buffer){
      //有buffer的情況下
      this._write(buffer.chunk,buffer.encoding,()=>this.clearBuffer(),buffer.callback)
    }else{
      //沒有的話 先看看需不需要drain
      if(this.needDrain){
        //觸發drain 並初始化所有狀態
        this.writing = false
        this.needDrain = false
        this.callbackList.shift()()
        this.emit('drain')

      }
      this.callbackList.map(v=>{
        v()
      })
      this.callbackList.length = 0
    }
  }
  _write(chunk,encoding,clearBuffer,callback){
    //因為write方法是同步調用的 所以可能還沒獲取到fd
    if(typeof this.fd != 'number'){
      //直接在open的時間對象上注冊一個一次性事件 當open被emit的時候會被調用
      return this.once('open',()=>this._write(chunk,encoding,clearBuffer,callback))
    }
    fs.write(this.fd,chunk,0,chunk.length,this.pos,(err,byteWrite)=>{
      this.pos += byteWrite
      //每次寫完 相應減少內存中的數量
      this.len -= byteWrite
      if(callback) this.callbackList.push(callback)
      //第一次寫完
      clearBuffer()

    })
  }

  //寫入方法
  write(chunk,encoding=this.encoding,callback){
    //判斷chunk必須是字符串或者buffer 為了統一都變成buffer
    chunk = Buffer.isBuffer(chunk)?chunk:Buffer.from(chunk,encoding)
    //維護緩存的長度 3
    this.len += chunk.length
    let ret = this.len < this.highWaterMark
    if(!ret){
      //表示要觸發drain事件
      this.needDrain = true
    }
    //正在寫入的應該放到內存中
    if(this.writing){
      this.cache.push({
        chunk,
        encoding,
        callback
      })
    }else{
      //這里是第一次寫的時候
      this.writing = true
      //專門實現寫的方法
      this._write(chunk,encoding,()=>this.clearBuffer(),callback)
    }
    // console.log(ret)
    //能不能繼續寫了 false代表下次寫的時候更占內存
    return ret
  }

  destory(){
    if(typeof this.fd != 'number'){
      return this.emit('close')
    }
    //如果文件被打開過 就關閉文件並且觸發close事件
    fs.close(this.fd,()=>{
      this.emit('close')
    })
  }
  open(){
    //fd表示的就是當前this.path的這個文件，從3開始(number類型)
    fs.open(this.path,this.flags,(err,fd)=>{
      //有可能fd這個文件不存在 需要做處理
      if(err){
        //如果有自動關閉 則幫他銷毀
        if(this.autoClose){
          //銷毀（關閉文件，出發關閉文件事件）
          this.destory()
        }
        //如果有錯誤 就會觸發error事件
        this.emit('error',err)
        return
      }
      //保存文件描述符
      this.fd = fd
      //當文件打開成功時觸發open事件
      this.emit('open',this.fd)
    })
  }
}

 

自定義可寫流

因為createWriteStream內部調用了WriteStream類，WriteStream又實現了Writable接口,WriteStream實現了_write()方法,所以我們通過自定義一個類繼承stream模塊的Writable，並在原型上自定義一個_write()就可以自定義自己的可寫流

let { Writable } = require('stream');

class MyWrite extends Writable{
  _write(chunk,encoding,callback){
    //write()的第一個參數，寫入的數據
    console.log(chunk);
    //這個callback，就相當於我們上面的clearBuffer方法，如果不執行callback就不會繼續從緩存中取出寫
    callback();
  }
}

let writer = new MyWrite();
writer.write('yes',()=>{
  console.log('ok');
});

返回：

node test.js
<Buffer 79 65 73>
ok

 

 

《2》可讀流

可讀流是對提供數據的來源的一種抽象。

可讀流的例子包括：

所有可讀流都實現了 stream.Readable 類定義的接口。

 

兩種讀取模式

可讀流運作於兩種模式之一：流動模式（flowing）或暫停模式（paused）。

  • 在流動模式中，數據自動從底層系統讀取，並通過 EventEmitter 接口的事件盡可能快地被提供給應用程序。
  • 在暫停模式中，必須顯式調用 stream.read() 讀取數據塊。

所有可讀流都開始於暫停模式，可以通過以下方式切換到流動模式：

可讀流可以通過以下方式切換回暫停模式：

  • 如果沒有管道目標，則調用 stream.pause()。
  • 如果有管道目標，則移除所有管道目標。調用 stream.unpipe() 可以移除多個管道目標。

只有提供了消費或忽略數據的機制后，可讀流才會產生數據。 如果消費的機制被禁用或移除，則可讀流會停止產生數據。

為了向后兼容，移除 'data' 事件句柄不會自動地暫停流。 如果有管道目標，一旦目標變為 drain 狀態並請求接收數據時，則調用 stream.pause() 也不能保證流會保持暫停模式。

如果可讀流切換到流動模式，且沒有可用的消費者來處理數據，則數據將會丟失。 例如，當調用 readable.resume() 時，沒有監聽 'data' 事件或 'data' 事件句柄已移除。

添加 'readable' 事件句柄會使流自動停止流動，並通過 readable.read() 消費數據。 如果 'readable' 事件句柄被移除，且存在 'data' 事件句柄，則流會再次開始流動。

 

三種狀態

可讀流的兩種模式是對發生在可讀流中更加復雜的內部狀態管理的一種簡化的抽象。

在任意時刻，可讀流會處於以下三種狀態之一：

  • readable.readableFlowing === null
  • readable.readableFlowing === false
  • readable.readableFlowing === true

當 readable.readableFlowing 為 null 時，沒有提供消費流數據的機制，所以流不會產生數據。 在這個狀態下，監聽 'data' 事件、調用 readable.pipe()、或調用 readable.resume() 都會使 readable.readableFlowing 切換到 true，可讀流開始主動地產生數據並觸發事件。

調用 readable.pause()、readable.unpipe()、或接收到背壓，則 readable.readableFlowing 會被設為 false，暫時停止事件流動但不會停止數據的生成。 在這個狀態下，為 'data' 事件綁定監聽器不會使 readable.readableFlowing 切換到 true。

const { PassThrough, Writable } = require('stream'); const pass = new PassThrough(); const writable = new Writable(); pass.pipe(writable); pass.unpipe(writable); // readableFlowing 現在為 false。  pass.on('data', (chunk) => { console.log(chunk.toString()); }); pass.write('ok'); // 不會觸發 'data' 事件。 pass.resume(); // 必須調用它才會觸發 'data' 事件。如果注釋掉它則不會返回結果ok

當 readable.readableFlowing 為 false 時，數據可能會堆積在流的內部緩沖中。

 

選擇一種接口風格

可讀流的 API 貫穿了多個 Node.js 版本，且提供了多種方法來消費流數據。 ⚠️開發者通常應該選擇其中一種方法來消費數據，不要在單個流使用多種方法來消費數據。 混合使用 on('data')、on('readable')、pipe() 或異步迭代器，會導致不明確的行為。

對於大多數用戶，建議使用 readable.pipe()，因為它是消費流數據最簡單的方式。 如果開發者需要精細地控制數據的傳遞與產生，可以使用 EventEmitter、readable.on('readable')/readable.read() 或 readable.pause()/readable.resume()。

 

stream.Readable 類

下面是事件的介紹：

'error' 事件

當流因底層內部出錯而不能產生數據、或推送無效的數據塊時觸發。

'close' 事件

當流或其底層資源（比如文件描述符）被關閉時觸發。 表明不會再觸發其他事件，也不會再發生操作。

不是所有可讀流都會觸發 'close' 事件。

'data' 事件
  • chunk <Buffer> | <string> | <any> 數據塊。 對於非對象模式的流， chunk 可以是字符串或 Buffer。 對於對象模式的流，chunk 可以是任何 JavaScript 值，除了 null。

當流將數據塊傳送給消費者后觸發。 當調用 readable.pipe()，readable.resume() 或綁定監聽器到 'data' 事件時，流會轉換到流動模式。 當調用 readable.read() 且有數據塊返回時，也會觸發 'data' 事件。

如果使用 readable.setEncoding() 為流指定了默認的字符編碼，則監聽器回調傳入的數據為字符串，否則傳入的數據為 Buffer。

const fs = require('fs'); const rr = fs.createReadStream('data.txt');//hello data rr.on('data', (chunk) => {//readable不行,報錯TypeError: Cannot read property 'length' of undefined console.log(`接收到 ${chunk.length} 個字節的數據`); //chunk為undefined });

返回：

node test.js
接收到 10 個字節的數據

 

process.stdin.setEncoding('utf8'); process.stdin.on('data', (chunk) => {//readable不行，會閃退？？？？？？？ console.log(`接收到 ${chunk.length} 個字節的數據`); });

返回：

node test.js
今天天氣好 //自己輸入並回車，這個內容就會被process.stdin收到 接收到 6 個字節的數據

 

之前有試一個例子一直沒有成功：

process.stdin.setEncoding('utf8'); // process.stdout.write("請輸入用戶名:"); process.stdin.on('data', (chunk) => { // var chunk = process.stdin.read();  console.log(chunk); if (chunk !== null) { process.stdout.write(`data: ${chunk}`); } }); process.stdin.on('end', () => { process.stdout.write('end'); });

 

'end' 事件

當流中沒有數據可供消費時觸發。

'end' 事件只有在數據被完全消費掉后才會觸發。 要想觸發該事件，可以將流轉換到流動模式，或反復調用 stream.read() 直到數據被消費完。

'readable' 事件

當流中有數據可供讀取時觸發

當到達流數據尾部時， 'readable' 事件也會觸發。觸發順序在 'end' 事件之前。

事實上， 'readable' 事件表明流有了新的動態：要么是有了新的數據，要么是到了流的尾部。 對於前者， stream.read() 將返回可用的數據。而對於后者， stream.read() 將返回 null。 例如，下面的例子中的 foo.txt 是一個空文件：

const fs = require('fs'); const rr = fs.createReadStream('foo.txt'); rr.on('readable', () => { console.log(`讀取的數據: ${rr.read()}`); }); rr.on('end', () => { console.log('結束'); });

返回：

node test.js
讀取的數據: null 結束

 

有問題：

const fs = require('fs'); const rr = fs.createReadStream('data.txt'); rr.on('readable', function(){//不能是'data'事件，為什么，如果是data，返回只有null和end，明天好好查查這兩者的對比 var chunk = rr.read(); // 獲取到輸入的信息  console.log(chunk); if(chunk === ''){ rr.emit('end'); // 觸發end事件 return } if (chunk !== null) { process.stdout.write('data: '+ chunk +'\n'); } // rr.emit('end'); }); rr.on('end', function() { process.stdout.write('end'+'\n'); //也輸出了，只是被擋住了，加上+'\n'就看出來了 });

返回：

node test.js
<Buffer 68 65 6c 6c 6f 20 64 61 74 61> data: hello data null end

 上面標明的錯誤都是因為一開始沒能弄清楚data和readable的區別，看了博客https://blog.csdn.net/eeewwwddd/article/details/81042225?utm_source=copy后終於明白


參數

path:讀取的文件的路徑

option:
highWaterMark:水位線，一次可讀的字節，一般默認是64k
flags:標識，打開文件要做的操作，默認是r
encoding:編碼，默認為buffer
start:開始讀取的索引位置
end:結束讀取的索引位置(包括結束位置)
autoClose:讀取完畢是否關閉，默認為true

 

data與readable的區別：

  • readable和讀流的data的區別就是，readable可以控制自己從緩存區讀多少和控制讀的次數，而data是每次讀取都清空緩存，讀多少輸出多少
  • readable是暫停模式，data是流動模式；就是readable需要使用read()來讀取數據，data則是從回調中就能夠得到數據
const fs = require('fs');
//讀取的時候默認讀64k
let rs = fs.createReadStream('./data.txt',{//內容為hello data
  highWaterMark: 2,//一次讀的字節 默認64k
  flags: 'r',      //標示 r為讀 w為寫
  autoClose: true, //默認讀取完畢后自動關閉
  start: 0,
  end: 5,          //流是閉合區間包start，也包end 默認是讀完
  encoding: 'utf8' //默認編碼是buffer
});
rs.on('data',(data) => {
    console.log('data');
    console.log(data);
});
//因為上面的data事件把數據讀了，清空緩存區。所以導致下面的readable讀出為null
rs.on('readable',() => {
    console.log('readable');
    console.log(rs.read());
});

返回：

node test.js
data
he
data
ll
data
o 
readable
null

如果把'data'監聽去掉，那么返回結果就是：

node test.js
readable
he
readable
ll
readable
o 
readable
null

 

舉例說明readable的使用情況：

（1）

let rs = fs.createReadStream('./foo.txt', {//內容為 Today is a good day.i want to go out for fun.
  //每次讀7個
  highWaterMark: 7,
  encoding: 'utf8'
})
//如果讀流第一次全部讀下來並且小於highWaterMark，就會再讀一次(再觸發一次readable事件)
rs.on('readable', () => {
  let result = rs.read(2);
  console.log(result)
})

返回：

node test.js
To
da

（2）

//如果rs.read()不加參數，一次性讀完，會從緩存區再讀一次，為null
rs.on('readable', () => {
  let result = rs.read();
  console.log(result)
})

返回：

node test.js
Today i
s a goo
d day.i
 want t
o go ou
t for f
un.
null

（3）

//如果readable每次都剛好讀完(即rs.read()的參數剛好和highWaterMark相等)，就會一直觸發readable事件，如果最后不足他想喝的數，他就會先觸發一次null，最后把剩下的喝完
rs.on('readable', () => {
  let result = rs.read();
  console.log(result)
})

返回：

node test.js
Today i
s a goo
d day.i
 want t
o go ou
t for f
null
un.

（4）

//一開始緩存區為0的時候也會默認調一次readable事件，將foo.txt內容清零
rs.on('readable', () => {
  let result = rs.read();
  console.log(result)
})

返回：

node test.js
null

 

實戰：行讀取器(平常我們的文件可能有回車、換行，此時如果要每次想讀一行的數據，就得用到readable)

let EventEmitter = require('events')
//如果要將內容全部讀出就用on('data')，精確讀取就用on('readable')
class LineReader extends EventEmitter {
  constructor(path) {
    super()
    this.rs = fs.createReadStream(path)
    //回車符的十六進制
    let RETURN = 0x0d
    //換行符的十六進制
    let LINE = 0x0a
    let arr = []
    this.on('newListener', (type) => {//每次使用 on 監聽事件時觸發'newListener'事件
      if (type === 'newLine') {//自定義的一個事件'newLine'，觸發后就調用'readable'，然后自行設定一次讀取一行的操作
        this.rs.on('readable', () => {
          let char
          //每次讀一個，當讀完的時候會返回null，終止循環
          while (char = this.rs.read(1)) {//讀到文件最后char = null
            switch (char[0]) {
              case RETURN:
                break;
              //Mac下只有換行符，windows下是回車符和換行符，需要根據不同的轉換。因為我這里是Mac
              case LINE:
                //如果是換行符就把數組轉換為字符串
                let r = Buffer.from(arr).toString('utf8')
                //把數組清空
                arr.length = 0
                //觸發newLine事件，把得到的一行數據輸出
                this.emit('newLine', r)
                break;
              default:
                //如果不是換行符，就放入數組中
                arr.push(char[0])
            }
          }
        })
      }
    })
    //以上只能取出換行符之前的代碼，最后一行的后面沒有換行符，所以需要特殊處理。當讀流讀完需要觸發end事件時
    this.rs.on('end', () => {
      //取出最后一行數據，轉成字符串
      let r = Buffer.from(arr).toString('utf8')
      arr.length = 0
      this.emit('newLine', r)
    })
  }
}

let lineReader = new LineReader('./foo.txt')
lineReader.on('newLine', function (data) {
  console.log('a line');
  console.log(data);
})

返回：

node test.js //可見一次是只讀取一行的
a line
if the truth is :
a line
I
a line
Am
a line
A
a line
boy

一般是將整個文件讀取完的：

const fs = require('fs');
const rr = fs.createReadStream('foo.txt',{encoding: 'utf8'});
rr.on('readable', () => {
  console.log('one time');
  console.log(rr.read());
});
rr.on('end', () => {
  console.log('結束');
});

返回：

node test.js
one time
if the truth is :
I
Am
A
boy
one time
null
結束

 

下面接着方法的介紹：

readable.destroy([error])

銷毀流，並且觸發error事件。然后，可讀流將釋放所有的內部資源。

開發者不應該覆蓋這個方法，應該覆蓋readable._destroy方法。

const fs = require('fs');
const rr = fs.createReadStream('foo.txt',{encoding: 'utf8'});
rr.on('open', function () {
    console.log('文件被打開');
});
rr.destroy('something wrong');//有參數則為出現的錯誤，會觸發error事件
rr.on('data', function (data) {
    console.log('data');
    console.log(data);

});
rr.on('error', function (err) {
    console.log('error');
    console.log(err);
});
rr.on('close', function (err) {
    console.log('close');
});
rr.on('end', () => {
  console.log('end');
});

返回：

node test.js
文件被打開
error
something wrong

如果rr.destroy();參數為空，則不會觸發error事件，而是觸發close事件，那么返回為：

 

node test.js
文件被打開
close
readable.isPaused()

readable.isPaused() 方法返回可讀流的當前操作狀態。 該方法主要是在 readable.pipe() 方法的底層機制中用到。大多數情況下，沒有必要直接使用該方法

readable.pause()
  • 返回： this

readable.pause() 方法將會使 flowing 模式的流停止觸發 'data' 事件， 進而切出 flowing 模式。任何可用的數據都將保存在內部緩存中。

 

readable.read([size])

readable.read()方法從內部緩沖區中抽出並返回一些數據。 如果沒有可讀的數據，返回null。readable.read()方法默認數據將作為“Buffer”對象返回 ，除非已經使用readable.setEncoding()方法設置編碼或流運行在對象模式。

可選的size參數指定要讀取的特定數量的字節。如果size字節不可讀，將返回null除非流已經結束，在這種情況下所有保留在內部緩沖區的數據將被返回。

如果沒有指定size參數，則內部緩沖區包含的所有數據將返回。

readable.read()方法只應該在暫停模式下的可讀流上運行。在流模式下，readable.read()自動調用直到內部緩沖區的數據完全耗盡。

一般來說，建議開發人員避免使用'readable'事件和readable.read()方法，使用readable.pipe()或'data'事件代替。

無論size參數的值是什么，對象模式中的可讀流將始終返回調用readable.read(size)的單個項目。

注意：如果readable.read()方法返回一個數據塊，那么一個'data'事件也將被發送。

注意：在已經被發出的'end'事件后調用stream.read([size])事件將返回null。不會拋出運行時錯誤。

 

//fd文件描述符，一般通過fs.open中獲取
//buffer是讀取后的數據放入的緩存目標
//0，從buffer的0位置開始放入
//BUFFER_SIZE，每次放BUFFER_SIZE這么長的長度
//index，每次從文件的index的位置開始讀
//bytesRead，真實讀到的個數
fs.read(fd,buffer,0,BUFFER_SIZE,index,function(err,bytesRead){

})

 

 

 

 

readable.resume()
  • 返回： this

readable.resume() 方法會重新觸發 'data' 事件, 將暫停模式切換到流動模式。

readable.resume() 方法可以用來充分使用流中的數據，而不用實際處理任何數據，如以下示例所示：

getReadableStreamSomehow()
  .resume()
  .on('end', () => {
    console.log('Reached the end, but did not read anything.');
  });
readable.setEncoding(encoding)
  • encoding <string> 要使用的編碼
  • Returns: this

readble.setEncoding() 方法會為從可讀流讀入的數據設置字符編碼

默認返回Buffer對象。設置編碼會使得該流數據返回指定編碼的字符串而不是Buffer對象。例如，調用readable.setEncoding('utf8')會使得輸出數據作為UTF-8數據解析，並作為字符串返回。調用readable.setEncoding('hex')使得數據被編碼成16進制字符串格式。

可讀流會妥善處理多字節字符，如果僅僅直接從流中取出Buffer對象，很可能會導致錯誤解碼。

 

 

舉例說明上面的事件和方法的使用：

const fs = require('fs');
const rr = fs.createReadStream('foo.txt',{encoding: 'utf8'});
rr.on('open', function () {//1 先響應open
    console.log('文件被打開');
});
rr.on('data', function (data) {//2 
    console.log('data');
    console.log(rr.isPaused()); //false
    rr.pause();//3 改為暫停模式，不讀取數據了
    console.log(rr.isPaused());//true
    console.log(data);

});
setTimeout(function () {//7 兩秒后恢復成流動模式繼續讀取數據
    console.log('resume');
    console.log(rr.isPaused());//true
    rr.resume();
    console.log(rr.isPaused());//true,因為添加 'readable' 事件句柄會使流自動停止流動，並通過 readable.read() 消費數據。 如果 'readable' 事件句柄被移除，且存在 'data' 事件句柄，則流會再次開始流動
},1000);                       //注釋掉readable后，結果就為false

rr.on('error', function (err) {
    console.log(err);
});
rr.on('readable', () => {//4 因為data將所有數據都讀完並將緩存清空，所以readable只輸出null
  console.log('readable');
  console.log(rr.read());
});
rr.on('close', function (err) {//6 關閉
    console.log('close');
});
rr.on('end', () => {//5 結束
  console.log('end');
});

返回：

node test.js
文件被打開
data
false
true
if the truth is :
I
Am
A
boy
readable
null
end
close
resume
true
true

注釋掉readable返回：

node test.js
文件被打開
data
false
true
if the truth is :
I
Am
A
boy
resume
true
false
end
close

 

readable.pipe(destination[, options])

readable.pipe() 綁定一個 [Writable][] 到 readable 上， 將可寫流自動切換到 flowing 模式並將所有數據傳給綁定的 [Writable][]。數據流將被自動管理。這樣，即使是可讀流較快，目標可寫流也不會超負荷（overwhelmed）。

下面例子將 readable 中的所有數據通過管道傳遞給名為 foo.txt 的文件：

const fs = require('fs');
const rr = fs.createReadStream('foo.txt',{encoding: 'utf8'});
rr.pipe(process.stdout);

返回：

node test.js
if the truth is :
I
Am
A
boy

可以在單個可讀流上綁定多個可寫流。

readable.pipe() 方法返回 目標流 的引用，這樣就可以對流進行鏈式地管道操作：

const fs = require('fs');
const zlib = require('zlib');
const rr = fs.createReadStream('foo.txt',{encoding: 'utf8'});
const z = zlib.createGzip();
const w = fs.createWriteStream('foo.txt.gz');
rr.pipe(z).pipe(w);
//運行后，文件夾中果然出現了一個壓縮文件

默認情況下，當源可讀流（the source Readable stream）觸發 'end' 事件時，目標流也會調用 stream.end() 方法從而結束寫入。要禁用這一默認行為， end 選項應該指定為 false， 這將使目標流保持打開， 如下面例子所示：

const fs = require('fs');
const rr = fs.createReadStream('foo.txt',{encoding: 'utf8'});
const writer = fs.createWriteStream('foo2.txt');
rr.pipe(writer,{end:false});
rr.on('end', () => {
  console.log('end reader');
});
setTimeout(function(){
    writer.write('請輸入num1的值：');
    writer.end();
},2000);

返回：

node test.js
end reader

且foo2.txt文件中內容為：

if the truth is :
I
Am
A
boy請輸入num1的值：

如果去掉{ end: false }，則出錯：

node test.js
end reader
events.js:167
      throw er; // Unhandled 'error' event
      ^

Error [ERR_STREAM_WRITE_AFTER_END]: write after end //這就是因為當源可讀流觸發 'end' 事件時，目標流也會調用 stream.end() 方法從而結束寫入

這里有一點要警惕，如果可讀流在處理時發生錯誤，目標可寫流 不會 自動關閉。 如果發生錯誤，需要 手動 關閉所有流以避免內存泄漏。

注意：不管對 process.stderr 和 process.stdout 指定什么選項，它們都是直到 Node.js 進程退出才關閉。

 

readable.unpipe([destination])

readable.unpipe() 方法將之前通過stream.pipe()方法綁定的流分離

如果 destination 沒有傳入, 則所有綁定的流都會被分離.

如果傳入 destination, 但它沒有被pipe()綁定過，則該方法不作為.

const readable = getReadableStreamSomehow();
const writable = fs.createWriteStream('file.txt');
// All the data from readable goes into 'file.txt',
// but only for the first second
readable.pipe(writable);
setTimeout(() => {
  console.log('Stop writing to file.txt');
  readable.unpipe(writable);
  console.log('Manually close the file stream');
  writable.end();
}, 1000);

 

readable源碼實現，轉自https://blog.csdn.net/eeewwwddd/article/details/81042225

let fs = require('fs')
let EventEmitter = require('events')
class ReadStream extends EventEmitter{
  constructor(path,options = {}){
    super()
    this.path = path
    this.highWaterMark = options.highWaterMark || 64*1024
    this.flags = options.flags || 'r'
    this.start = options.start || 0
    this.pos = this.start     //會隨着讀取的位置改變
    this.autoClose = options.autoClose || true
    this.end = options.end || null
    //默認null就是buffer
    this.encoding = options.encoding || null

    //參數的問題
    this.reading = false //非流動模式
    //創建個buffer用來存儲每次讀出來的數據
    this.buffers = []
    //緩存區長度
    this.len = 0
    //是否要觸發readable事件
    this.emittedReadable = false
    //觸發open獲取文件的fd標識符
    this.open()
    //此方法默認同步調用 每次設置on監聽事件時都會調用之前所有的newListener事件
    this.on('newListener',(type)=>{// 等待着他監聽data事件
      if(type === 'readable'){
        //開始讀取 客戶已經監聽的data事件
        this.read()
      }
    })
  }
  //readable真正的源碼中的方法，計算出和n最接近的2的冪次數
  computeNewHighWaterMark(n) {
    n--;
    n |= n >>> 1;
    n |= n >>> 2;
    n |= n >>> 4;
    n |= n >>> 8;
    n |= n >>> 16;
    n++;
    return n;
  }
  read(n){
    //當讀的數量大於水平線，會通過取2的冪次取比他大和最接近的數
    if(this.len < n){
      this.highWaterMark = this.computeNewHighWaterMark(n)
      //重新觸發readbale的callback，所以第一次會觸發null
      this.emittedReadable = true
      //重新讀新的水位線
      this._read()
    }
    //真正讀取到的
    let buffer = null
    //說明緩存里有這么多，取出來
    if(n>0 && n<=this.len){
      //定義一個buffer
      buffer = Buffer.alloc(n)
      let buf
      let flag = true
      let index = 0
      //[buffer<1,2,3,4>,buffer<1,2,3,4>,buffer<1,2,3,4>]
      //每次取出緩存前的第一個buffer
      while(flag && (buf = this.buffers.shift())){
        for(let i=0;i<buf.length;i++){
          //把取出的一個buffer中的數據放入新定義的buffer中
          buffer[index++] = buf[i]
          //當buffer的長度和n(參數)長度一樣時，停止循環
          if(index === n){
            flag = false
            //維護緩存，因為可能緩存中的buffer長度大於n，當取出n的長度時，還會剩下其余的buffer，我們需要切割buf並且放到緩存數組之前
            this.len -= n
            let r = buf.slice(i+1)
            if(r.length){
              this.buffers.unshift(r)
            }
            break
          }
        }
      }
    }
    //如果緩存區沒有東西，等會讀完需要觸發readable事件
    //這里會有一種狀況，就是如果每次Readable讀取的數量正好等於highWaterMark(流讀取到緩存的長度)，就會每次都等於0，每次都觸發Readable事件，就會每次讀，讀到沒有為止，最后還會觸發一下null
    if(this.len === 0){
      this.emittedReadable = true
    }
    if(this.len < this.highWaterMark){
      //默認，一開始的時候開始讀取
      if(!this.reading){
        this.reading = true
        //真正多讀取操作
        this._read()
      }
    }
    return buffer&&buffer.toString()
  }
  _read(){
    if(typeof this.fd != 'number'){
      //等待着觸發open事件后fd肯定拿到了 再去執行read方法
      return this.once('open',()=>{this._read()})
    }
    //先讀這么多buffer
    let buffer = Buffer.alloc(this.highWaterMark)
    fs.read(this.fd,buffer,0,buffer.length,this.pos,(err,byteRead)=>{
      if(byteRead > 0){
        //當第一次讀到數據后，改變reading的狀態，如果觸發read事件，可能還會在觸發第二次_read
        this.reading = false
        //每次讀到數據增加緩存取得長度
        this.len += byteRead
        //每次讀取之后，會增加讀取的文件的讀取開始位置
        this.pos += byteRead
        //將讀到的buffer放入緩存區buffers中
        this.buffers.push(buffer.slice(0,byteRead))
        //觸發readable
        if(this.emittedReadable){
          this.emittedReadable = false
          //可以讀取了，默認開始的時候杯子填滿了
          this.emit('readable')
        }
      }else{
        //沒讀到就出發end事件
        this.emit('end')
      }
    })
  }
  destory(){
    if(typeof this.fd != 'number'){
      return this.emit('close')
    }
    //如果文件被打開過 就關閉文件並且觸發close事件
    fs.close(this.fd,()=>{
      this.emit('close')
    })
  }
  open(){
    //fd表示的就是當前this.path的這個文件，從3開始(number類型)
    fs.open(this.path,this.flags,(err,fd)=>{
      //有可能fd這個文件不存在 需要做處理
      if(err){
        //如果有自動關閉 則幫他銷毀
        if(this.autoClose){
          //銷毀（關閉文件，觸發關閉文件事件）
          this.destory()
        }
        //如果有錯誤 就會觸發error事件
        this.emit('error',err)
        return
      }
      //保存文件描述符
      this.fd = fd
      //當文件打開成功時觸發open事件
      this.emit('open',this.fd)
    })
  }
}

 

自定義可讀流

因為createReadStream內部調用了ReadStream類，ReadStream又實現了Readable接口,ReadStream實現了_read()方法,所以我們通過自定義一個類繼承stream模塊的Readable，並在原型上自定義一個_read()就可以自定義自己的可讀流

let { Readable } = require('stream');

class MyRead extends Readable{
  //流需要一個_read方法，方法中push什么，外面就接收什么
  _read(){
    //push方法就是上面_read方法中的push一樣，把數據放入緩存區中
    this.push('100');
    //如果push了null就表示沒有東西可讀了，停止(如果不寫，就會一直push上面的值，死循環)
    this.push(null);
  }
}

let reader = new MyRead({encoding:'utf8'});
reader.on('readable',() => {
    console.log(reader.read());
});

返回：

node test.js
100

 

 

pipe——管道 可以控制速率，因為讀快寫慢

let fs = require('fs')
//pipe方法叫管道 可以控制速率
let rs = fs.createReadStream('./foo.txt',{
  highWaterMark: 4
})
let ws = fs.createWriteStream('./foo1.txt',{
  highWaterMark: 1
})
//會監聽rs的on('data')將讀取到的數據,通過ws.write的方法寫入文件
//調用寫的一個方法 返回boolean類型
//如果返回false就調用rs的pause方法 暫停讀取
//等待可寫流 寫入完畢在監聽drain resume rs
rs.pipe(ws) //會控制速率 防止淹沒可用內存

 

let fs = require('fs')
//這兩個是上面自己寫的ReadStream和WriteStream
let { Readable } = require('stream');

class MyRead extends Readable{
  //流需要一個_read方法，方法中push什么，外面就接收什么
  _read(){
    //push方法就是上面_read方法中的push一樣，把數據放入緩存區中
    this.push('100');
    //如果push了null就表示沒有東西可讀了，停止(如果不寫，就會一直push上面的值，死循環)
    this.push(null);
  }
}

let writer = fs.createWriteStream('./foo1.txt',{
  highWaterMark: 1
});

//如果用原來的讀寫，因為寫比較耗時，所以會多讀少寫，耗內存
MyRead.prototype.pipe = function(dest){
  this.on('data',(data)=>{
    let flag = dest.write(data)
    //如果寫入的時候嘴巴吃滿了就不繼續讀了，暫停
    if(!flag){
      this.pause()
    }
  });
  //如果寫的時候嘴巴里的吃完了，就會繼續讀
  dest.on('drain',()=>{
    this.resume()
  });
  this.on('end',()=>{
    this.destroy()//銷毀ReadStream
    //清空緩存中的數據
    fs.fsync(1,()=>{//fs.fsync作用是同步磁盤緩存，1代表的是文件描述符，0，1，2 文件描述符代表標准輸入設備（比如鍵盤），標准輸出設備（顯示器）和標准錯誤
      dest.destroy()//銷毀WriteStream,之前dest設的是，但是報錯process.stdout cannot be closed
    });
  });
}
var reader = new MyRead();
reader.pipe(writer);//結果就是將100寫到了文件foo1.txt中

上面的文件描述符處本來寫的是dest.fd,但是報錯：

TypeError [ERR_INVALID_ARG_TYPE]: The "fd" argument must be of type number. Received type object

查看writer的fd為null，不知原因，待查明？？？？？？？？？？？

 

stream.pipeline(...streams[, callback])

  • ...streams <Stream> 兩個或多個要用管道連接的流
  • callback <Function> 一個回調函數，可以帶有一個錯誤信息參數

該模塊方法用於在多個流之間架設管道，可以自動傳遞錯誤和完成掃尾工作，並且可在管道架設完成時提供一個回調函數：

const { pipeline } = require('stream');
const fs = require('fs');
const zlib = require('zlib');

// 使用 pipeline API 輕松連接多個流
// 並在管道完成時獲得通知

// 使用pipeline高效壓縮一個可能很大的tar文件：

pipeline(
  fs.createReadStream('foo.txt'),
  zlib.createGzip(),
  fs.createWriteStream('foo.tar.gz'),//運行后成功壓縮並返回 管道架設成功 信息
  (err) => {
    if (err) {
      console.error('管道架設失敗', err);
    } else {
      console.log('管道架設成功');
    }
  }
);

pipeline API 也可做成承諾：

const util = require('util');
const stream = require('stream');
const fs = require('fs');
const zlib = require('zlib');
const pipeline = util.promisify(stream.pipeline);

async function run() {
  await pipeline(
    fs.createReadStream('foo.txt'),
    zlib.createGzip(),
    fs.createWriteStream('foo.tar.gz')////運行后成功壓縮並返回 管道架設成功 信息
  );
  console.log('管道架設成功');
}

run().catch(console.error);

 

 

用於實現流的 API

其實就是覆寫下面的這些方法來實現自己的流操作：

新的流類必須實現一個或多個特定的方法，根據所創建的流類型，如下圖所示:

用例 類 實現的方法
只讀流 Readable _read
只寫流 writable _write ，_writev，_final
可讀可寫流 Duplex _read ，_write ，_writev，_final
操作寫數據，然后讀結果 Transform _transform，_flush，_final

注意：實現流的代碼里面不應該出現調用“public”方法的地方因為這些方法是給使用者使用的（流使用者部分的API所述）。這樣做可能會導致使用流的應用程序代碼產生不利的副作用。

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

class MyWritable extends Writable {
  constructor(options) {
    super(options);
    // ...
  }
}

 

 

雙工流

有了雙工流，我們可以在同一個對象上同時實現可讀和可寫，就好像同時繼承這兩個接口。 重要的是雙工流的可讀性和可寫性操作完全獨立於彼此。這僅僅是將兩個特性組合成一個對象。

let { Duplex } = require('stream')
//雙工流，可讀可寫
class MyDuplex extends Duplex{
  _read(){
    this.push('hello Duplex')
    this.push(null)
  }
  _write(chunk,encoding,clearBuffer){
    console.log(chunk)
    clearBuffer()
  }
}

let myDuplex = new MyDuplex()
//process.stdin是node自帶的process進程中的可讀流，會監聽命令行的輸入
//process.stdout是node自帶的process進程中的可寫流，會監聽並輸出在命令行中
//所以這里的意思就是在命令行先輸出hello，然后我們輸入什么他就出來對應的buffer(先作為可讀流出來)
process.stdin.pipe(myDuplex).pipe(process.stdout)

返回：

node test.js
hello Duplex

 

 

轉換流

在讀寫過程中可以修改或轉換數據的 Duplex 流（例如 zlib.createDeflate()）

轉換流的輸出是從輸入中計算出來的。對於轉換流，我們不必實現read或write的方法，我們只需要實現一個transform方法，將兩者結合起來。它有write方法的意思，我們也可以用它來push數據。

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

class MyTransform extends Transform{
  _transform(chunk,encoding,callback){//5 myTransform2 push時則觸發myTransform的_transform
    console.log(chunk.toString().toUpperCase());//6 然后輸出from MyTransform2的大寫內容
    callback();
  }
}
let myTransform = new MyTransform();


class MyTransform2 extends Transform{
  _transform(chunk,encoding,callback){//2 觸發myTransform2的_transform
    console.log(chunk.toString().toUpperCase());//3 輸出input的大寫內容INPUT
    this.push('from MyTransform2');//4 將from MyTransform2內容寫入myTransform
    this.push(null);
    callback();
  }
}
let myTransform2 = new MyTransform2();

//此時myTransform2被作為可寫流觸發_transform，輸出輸入的大寫字符后，會通過可讀流push字符到下一個轉換流中
//當寫入的時候才會觸發transform的值，此時才會push，所以后面的pipe拿到的chunk是前面的push的值
process.stdin.pipe(myTransform2).pipe(myTransform);

返回：

node test.js
input //1 輸入回車
INPUT

FROM MYTRANSFORM2

 

總結

可讀流
在 flowing 模式下， 可讀流自動從系統底層讀取數據，並通過 EventEmitter 接口的事件盡快將數據提供給應用。
在 paused 模式下，必須顯式調用 stream.read() 方法來從流中讀取數據片段。
所有初始工作模式為 paused 的 Readable 流，可以通過下面三種途徑切換到 flowing 模式：

  • 監聽 ‘data’ 事件
  • 調用 stream.resume() 方法
  • 調用 stream.pipe() 方法將數據發送到 Writable

可讀流可以通過下面途徑切換到 paused 模式：

  • 如果不存在管道目標（pipe destination），可以通過調用 stream.pause() 方法實現。
  • 如果存在管道目標，可以通過取消 ‘data’ 事件監聽，並調用 stream.unpipe() 方法移除所有管道目標來實現。


可寫流
需要知道只有在嘴真正的吃滿了，並且等到把嘴里的和地上的饅頭(緩存中的)都吃下了才會觸發drain事件
第一次寫入會直接寫入文件中，后面會從緩存中一個個取


雙工流
只是對可寫可讀流的一種應用，既可作為可讀流，也能作為可寫流，並且作為可讀或者可寫時是隔離的


轉換流
一般轉換流是邊輸入邊輸出的，而且一般只有觸發了寫入操作時才會進入_transform方法中。跟雙工流的區別就是，他的可讀可寫是在一起的

 


免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。



猜您在找 NodeJS Stream 二：什么是 Stream nodejs中Stream的理解 nodeJS之流stream NodeJS stream 一：Buffer nodejs stream & buffer 互相轉換 NodeJS Stream 四：Writable NodeJS Stream 五：雙工流 NodeJS Stream 三：readable 理解 nodeJS 中的 buffer，stream 理解nodejs中的stream(流)
 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM