大熊君大話NodeJS之------Stream子產品

一，開篇分析

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

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

在unix裡，Stream是通過 

"|" 

實作的。在node中，作為内置的stream子產品，很多核心子產品和三方子產品都使用到。

和unix一樣，node stream主要的操作也是

.pipe()

，使用者可以使用反壓力機制來控制讀和寫的平衡。 

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

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

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

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

運作後會有亂碼出現，原因就是沒有設定指定的字元集，比如：“utf-8” 。

修改一下就好：

1  var http = require('http') ;
2  var server = http.createServer(function(req,res){
3     res.writeHeader(200,{
4         'Content-Type' : 'text/plain;charset=utf-8'  // 添加charset=utf-8
5     }) ;
6     res.end("Hello,大熊！") ;
7  }) ;
8  server.listen(8888) ;
9  console.log("http server running on port 8888 ...") ;      

運作結果：

為什麼使用Stream

node中的I/O是異步的，是以對磁盤和網絡的讀寫需要通過回調函數來讀取資料，下面是一個檔案下載下傳例子

上代碼：

1 var http = require('http') ;
2 var fs = require('fs') ;
3 var server = http.createServer(function (req, res) {
4     fs.readFile(__dirname + '/data.txt', function (err, data) {
5         res.end(data);
6     }) ;
7 }) ;
8 server.listen(8888) ;      

代碼可以實作需要的功能，但是服務在發送檔案資料之前需要緩存整個檔案資料到記憶體，如果"data.txt"檔案很 

大并且并發量很大的話，會浪費很多記憶體。因為使用者需要等到整個檔案緩存到記憶體才能接受的檔案資料，這樣導緻 

使用者體驗相當不好。不過還好

(req,res)

兩個參數都是Stream，這樣我們可以用

fs.createReadStream()

代替

fs.readFile()。如下：

1 var http = require('http') ;
2 var fs = require('fs') ;
3 var server = http.createServer(function (req, res) {
4     var stream = fs.createReadStream(__dirname + '/data.txt') ;
5     stream.pipe(res) ;
6 }) ;
7 server.listen(8888) ;      

.pipe()

方法監聽

fs.createReadStream()

的

'data'

 和

'end'

事件，這樣"data.txt"檔案就不需要緩存整 

個檔案，當用戶端連接配接完成之後馬上可以發送一個資料塊到用戶端。使用

.pipe()

另一個好處是可以解決當客戶 

端延遲非常大時導緻的讀寫不平衡問題。

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

二，執行個體引入

當記憶體中無法一次裝下需要處理的資料時，或者一邊讀取一邊處理更加高效時，我們就需要用到資料流。NodeJS中通過各種

Stream

來提供對資料流的操作。

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

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

代碼中

data

事件會源源不斷地被觸發，不管

doSomething

函數是否處理得過來。代碼可以繼續做如下改造，以解決這個問題。

1 var rs = fs.createReadStream(src) ;
 2 rs.on('data', function (chunk) {
 3     rs.pause() ;
 4     doSomething(chunk, function () {
 5         rs.resume() ;
 6     }) ;
 7 }) ;
 8 rs.on('end', function () {
 9     cleanUp();
10 })  ;      

給

doSomething

函數加上了回調，是以我們可以在處理資料前暫停資料讀取，并在處理資料後繼續讀取資料。

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

1 var rs = fs.createReadStream(src) ;
2 var ws = fs.createWriteStream(dst) ;
3 rs.on('data', function (chunk) {
4     ws.write(chunk);
5 }) ;
6 rs.on('end', function () {
7     ws.end();
8 }) ;      

doSomething

換成了往隻寫資料流裡寫入資料後，以上代碼看起來就像是一個檔案拷貝程式了。但是以上代碼存在上邊提到的問題，如果寫入速度跟不上讀取速度的話，隻寫資料流内部的緩存會爆倉。我們可以根據

.write

方法的傳回值來判斷傳入的資料是寫入目标了，還是臨時放在了緩存了，并根據

drain

事件來判斷什麼時候隻寫資料流已經将緩存中的資料寫入目标，可以傳入下一個待寫資料了。是以代碼如下：

1 var rs = fs.createReadStream(src) ;
 2 var ws = fs.createWriteStream(dst) ;
 3 rs.on('data', function (chunk) {
 4     if (ws.write(chunk) === false) {
 5         rs.pause() ;
 6     }
 7 }) ;
 8 rs.on('end', function () {
 9     ws.end();
10 });
11 ws.on('drain', function () {
12     rs.resume();
13 }) ;      

最終實作了資料從隻讀資料流到隻寫資料流的搬運，并包括了防爆倉控制。因為這種使用場景很多，例如上邊的大檔案拷貝程式，NodeJS直接提供了

.pipe

方法來做這件事情，其内部實作方式與上邊的代碼類似。

下面是一個更加完整的複制檔案的過程：

1 var fs = require('fs'),
 2   path = require('path'),
 3   out = process.stdout;
 4 
 5 var filePath = '/bb/bigbear.mkv';
 6 
 7 var readStream = fs.createReadStream(filePath);
 8 var writeStream = fs.createWriteStream('file.mkv');
 9 
10 var stat = fs.statSync(filePath);
11 
12 var totalSize = stat.size;
13 var passedLength = 0;
14 var lastSize = 0;
15 var startTime = Date.now();
16 
17 readStream.on('data', function(chunk) {
18 
19   passedLength += chunk.length;
20 
21   if (writeStream.write(chunk) === false) {
22     readStream.pause();
23   }
24 });
25 
26 readStream.on('end', function() {
27   writeStream.end();
28 });
29 
30 writeStream.on('drain', function() {
31   readStream.resume();
32 });
33 
34 setTimeout(function show() {
35   var percent = Math.ceil((passedLength / totalSize) * 100);
36   var size = Math.ceil(passedLength / 1000000);
37   var diff = size - lastSize;
38   lastSize = size;
39   out.clearLine();
40   out.cursorTo(0);
41   out.write('已完成' + size + 'MB, ' + percent + '%, 速度：' + diff * 2 + 'MB/s');
42   if (passedLength < totalSize) {
43     setTimeout(show, 500);
44   } else {
45     var endTime = Date.now();
46     console.log();
47     console.log('共用時：' + (endTime - startTime) / 1000 + '秒。');
48   }
49 }, 500);      

可以把上面的代碼儲存為 "

copy.js"

 試驗一下我們添加了一個遞歸的 

setTimeout

 （或者直接使用setInterval）來做一個旁觀者，

每500ms觀察一次完成進度，并把已完成的大小、百分比和複制速度一并寫到控制台上，當複制完成時，計算總的耗費時間。

三，總結一下

（1），了解Stream概念。

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

（3），注意細節的把控，比如：大檔案的拷貝，采用的使用 “chunk data” 的形式進行分片處理。

（4），pipe的使用

（5），再次強調一個概念：一個TCP連接配接既是可讀流，又是可寫流，而Http連接配接則不同，一個http request對象是可讀流，而http response對象則是可寫流。

 　　　　　　　　　　　　　　  哈哈哈，本篇結束，未完待續，希望和大家多多交流夠溝通，共同進步。。。。。。呼呼呼……(*^__^*)