Nodejs學習筆記之Stream模組_node.js

一，開頭分析

流是一個抽象接口，被 Node 中的許多物件所實現。例如對一個 HTTP 伺服器的請求是一個流，stdout 也是一個流。流是可讀，可寫或兼具兩者的。

最早接觸Stream是從早期的unix開始的， 數十年的實踐證明Stream 思想可以很簡單的開發出一些龐大的系統。

在unix裡，Stream是透過 "|" 實現的。在node中，作為內建的stream模組，許多核心模組和三方模組都使用到。

和unix一樣，node stream主要的操作也是.pipe()，使用者可以使用反壓力機制來控制讀取和寫入的平衡。

Stream 可以提供開發者可以重複使用統一的接口，透過抽象的Stream介面來控制Stream之間的讀寫平衡。

一個TCP連接既是可讀流，又是可寫流，而Http連接則不同，一個http request物件是可讀流，而http response物件則是可寫流。

流的傳輸過程預設是以buffer的形式傳輸的，除非你給他設定其他編碼形式,以下是一個例子：

複製程式碼 程式碼如下:

 var http = require('http') ;
  var server = http.createServer(function(req,res){
     res.writeHeader(200, {'Content-Type': 'text/plain'}) ;
     res.end("Hello,大熊！") ;
  }) ;
  server.listen(8888) ;
  console.log("http server running on port 8888 ...") ;

運行後會有亂碼出現，原因就是沒有設定指定的字元集，例如：「utf-8」 。

修改一下就好：

複製程式碼 程式碼如下:

 var http = require('http') ;
 var server = http.createServer(function(req,res){
    res.writeHeader(200,{
        'Content-Type' : 'text/plain;charset=utf-8'  // 新增charset=utf-8
    }) ;
    res.end("Hello,大熊！") ;
 }) ;
 server.listen(8888) ;
 console.log("http server running on port 8888 ...") ;

運行結果：

為什麼要使用Stream
node中的I/O是異步的，因此對磁碟和網路的讀寫需要透過回調函數來讀取數據，下面是一個檔案下載範例
上碼：

複製程式碼 程式碼如下:

 var http = require('http') ;
 var fs = require('fs') ;
 var server = http.createServer(function (req, res) {
     fs.readFile(__dirname '/data.txt', function (err, data) {
         res.end(data);
     }) ;
 }) ;
 server.listen(8888) ;

程式碼可以實現需要的功能，但是服務在發送文件資料之前需要緩存整個文件資料到內存，如果"data.txt"文件很
大且並發量很大的話，會浪費很多記憶體。因為使用者需要等到整個檔案快取到記憶體才能接受的檔案數據，這樣導致
用戶體驗相當不好。不過還好(req,res)兩個參數都是Stream，這樣我們可以用fs.createReadStream()來取代fs.readFile()。如下：

複製程式碼 程式碼如下:

 var http = require('http') ;
 var fs = require('fs') ;
 var server = http.createServer(function (req, res) {
     var stream = fs.createReadStream(__dirname '/data.txt') ;
     stream.pipe(res) ;
 }) ;
 server.listen(8888) ;

.pipe()方法監聽fs.createReadStream()的'data' 和'end'事件，這樣"data.txt"檔案就不需要快取整
個文件，當客戶端連線完成之後馬上可以發送一個資料塊到客戶端。使用.pipe()另一個好處是可以解決當客戶
端延遲非常大時導致的讀寫不平衡問題。

有五種基本的Stream：readable，writable，transform，duplex，and "classic」。 （具體使用請自己查閱api）

二，實例引入

當記憶體中無法一次裝下需要處理的資料時，或是一邊讀取一邊處理更有效率時，我們就需要用到資料流。 NodeJS中透過各種Stream來提供資料流的操作。

以大檔案拷貝程式為例，我們可以為資料來源建立一個唯讀資料流，範例如下：

複製程式碼 程式碼如下:

 var rs = fs.createReadStream(pathname);
 rs.on('data', function (chunk) {
     doSomething(chunk) ; // 具體細節自己任意發揮
 });
 rs.on('end', function () {
     cleanUp() ;
 }) ;

程式碼中data事件會源源不斷地被觸發，不管doSomething函數是否處理得過來。程式碼可以繼續做以下改造，以解決這個問題。

複製程式碼 程式碼如下:

 var rs = fs.createReadStream(src) ;
 rs.on('data', function (chunk) {
     rs.pause() ;
     doSomething(chunk, function () {
         rs.resume() ;
     }) ;
 }) ;
 rs.on('end', function () {
     cleanUp();
 })  ;

為doSomething函數加上了回調，因此我們可以在處理資料之前暫停資料讀取，並在處理資料後繼續讀取資料。

此外，我們也可以為資料目標建立一個只寫資料流，如下：

複製程式碼 程式碼如下:

 var rs = fs.createReadStream(src) ;
 var ws = fs.createWriteStream(dst) ;
 rs.on('data', function (chunk) {
     ws.write(chunk);
 }) ;
 rs.on('end', function () {
     ws.end();
 }) ;

doSomething換成了往只寫資料流裡寫入資料後，以上程式碼看起來就像是文件拷貝程式了。但是以上程式碼有上邊提到的問題，如果寫入速度跟不上讀取速度的話，只寫資料流內部的快取會爆倉。我們可以根據.write方法的回傳值來判斷傳入的資料是寫入目標了，還是暫時放在了快取了，並根據drain事件來判斷什麼時候只寫資料流已經將快取中的資料寫入目標，可以傳入下一個待寫資料了。因此程式碼如下：

複製程式碼 程式碼如下:

 var rs = fs.createReadStream(src) ;
 var ws = fs.createWriteStream(dst) ;
 rs.on('data', function (chunk) {
     if (ws.write(chunk) === false) {
         rs.pause() ;
     }
 }) ;
 rs.on('end', function () {
     ws.end();
 });
 ws.on('drain', function () {
     rs.resume();
 }) ;

最終實現了資料從唯讀資料流到只寫資料流的搬運，並包括了防爆倉控制。因為這種使用場景很多，例如上邊的大文件拷貝程序，NodeJS直接提供了.pipe方法來做這件事情，其內部實現方式與上邊的代碼類似。

以下是一個更完整的複製檔案的過程：

複製程式碼 程式碼如下:

var fs = require('fs'),
  路徑 = require('路徑'),
  out = process.stdout;
var filePath = '/bb/bigbear.mkv';
var readStream = fs.createReadStream(filePath);
var writeStream = fs.createWriteStream('file.mkv');
var stat = fs.statSync(filePath);
var TotalSize = stat.size;
var passLength = 0;
var 最後大小 = 0;
var startTime = Date.now();
readStream.on('data', function(chunk) {
  PassedLength = chunk.length;
  if (writeStream.write(chunk) === false) {
    readStream.pause();
  }
});
readStream.on('end', function() {
  writeStream.end();
});
writeStream.on('drain', function() {
  readStream.resume();
});
setTimeout(function show() {
  var% = Math.ceil((提交的長度/總大小) * 100);
  var size = Math.ceil(passedLength / 1000000);
  var diff = size - lastSize;
  最後大小 = 大小;
  out.clearLine();
  out.cursorTo(0);
  out.write('完成' size 'MB, '% '%, 速度：' diff * 2 'MB/s');
  if (passedLength     setTimeout(顯示, 500);
  }其他{
    var endTime = Date.now();
    console.log();
    console.log('消耗時間：' (endTime - startTime) / 1000 '秒。');
  }
}, 500);

可以把上面的程式碼儲存為「copy.js」試試看我們加入一個潛水的setTimeout（或直接使用setInterval）模擬一個旁觀者，

每500ms觀察完成一次細節，將已完成的大小、百分比和複製速度一併寫入到控制台上，當複製完成時，計算總的計數時間。

三，總結一下 （1），理解Stream概念。

（2），熟練使用相關Stream的api

（3），注意細節的把控，例如：大文件的拷貝，宜使用「chunk data」的形式進行分片處理。

（4），管道的使用

（5），再次強調一個概念：一個TCP連接既是可寫流，又是可寫流，而Http連接則不同，一個http請求對像是可寫流，而http響應對像是可寫流。