快捷導(dǎo)航

Nodejs學(xué)習(xí)筆記之Stream模塊

更新時(shí)間：2015年01月13日 10:30:40 投稿：hebedich

nodejs的fs模塊并沒(méi)有提供一個(gè)copy的方法，但我們可以很容易的實(shí)現(xiàn)一個(gè)，這種方式是把文件內(nèi)容全部讀入內(nèi)存，然后再寫(xiě)入文件，對(duì)于小型的文本文件，這沒(méi)有多大問(wèn)題，比如grunt-file-copy就是這樣實(shí)現(xiàn)的。

一，開(kāi)篇分析

流是一個(gè)抽象接口，被 Node 中的很多對(duì)象所實(shí)現(xiàn)。比如對(duì)一個(gè) HTTP 服務(wù)器的請(qǐng)求是一個(gè)流，stdout 也是一個(gè)流。流是可讀，可寫(xiě)或兼具兩者的。

最早接觸Stream是從早期的unix開(kāi)始的，數(shù)十年的實(shí)踐證明Stream 思想可以很簡(jiǎn)單的開(kāi)發(fā)出一些龐大的系統(tǒng)。

在unix里，Stream是通過(guò) "|" 實(shí)現(xiàn)的。在node中，作為內(nèi)置的stream模塊，很多核心模塊和三方模塊都使用到。

和unix一樣，node stream主要的操作也是.pipe()，使用者可以使用反壓力機(jī)制來(lái)控制讀和寫(xiě)的平衡。

Stream 可以為開(kāi)發(fā)者提供可以重復(fù)使用統(tǒng)一的接口，通過(guò)抽象的Stream接口來(lái)控制Stream之間的讀寫(xiě)平衡。

一個(gè)TCP連接既是可讀流，又是可寫(xiě)流，而Http連接則不同，一個(gè)http request對(duì)象是可讀流，而http response對(duì)象則是可寫(xiě)流。

流的傳輸過(guò)程默認(rèn)是以buffer的形式傳輸?shù)模悄憬o他設(shè)置其他編碼形式,以下是一個(gè)例子：

復(fù)制代碼代碼如下:

 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 ...") ;

運(yùn)行后會(huì)有亂碼出現(xiàn)，原因就是沒(méi)有設(shè)置指定的字符集，比如：“utf-8” 。

修改一下就好：

復(fù)制代碼代碼如下:

 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 ...") ;

運(yùn)行結(jié)果：

為什么使用Stream
node中的I/O是異步的，因此對(duì)磁盤(pán)和網(wǎng)絡(luò)的讀寫(xiě)需要通過(guò)回調(diào)函數(shù)來(lái)讀取數(shù)據(jù)，下面是一個(gè)文件下載例子
上代碼：

復(fù)制代碼代碼如下:

 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) ;

代碼可以實(shí)現(xiàn)需要的功能，但是服務(wù)在發(fā)送文件數(shù)據(jù)之前需要緩存整個(gè)文件數(shù)據(jù)到內(nèi)存，如果"data.txt"文件很
大并且并發(fā)量很大的話，會(huì)浪費(fèi)很多內(nèi)存。因?yàn)橛脩粜枰鹊秸麄€(gè)文件緩存到內(nèi)存才能接受的文件數(shù)據(jù)，這樣導(dǎo)致
用戶體驗(yàn)相當(dāng)不好。不過(guò)還好(req,res)兩個(gè)參數(shù)都是Stream，這樣我們可以用fs.createReadStream()代替fs.readFile()。如下：

復(fù)制代碼代碼如下:

 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()方法監(jiān)聽(tīng)fs.createReadStream()的'data' 和'end'事件，這樣"data.txt"文件就不需要緩存整
個(gè)文件，當(dāng)客戶端連接完成之后馬上可以發(fā)送一個(gè)數(shù)據(jù)塊到客戶端。使用.pipe()另一個(gè)好處是可以解決當(dāng)客戶
端延遲非常大時(shí)導(dǎo)致的讀寫(xiě)不平衡問(wèn)題。

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

二，實(shí)例引入

當(dāng)內(nèi)存中無(wú)法一次裝下需要處理的數(shù)據(jù)時(shí)，或者一邊讀取一邊處理更加高效時(shí)，我們就需要用到數(shù)據(jù)流。NodeJS中通過(guò)各種Stream來(lái)提供對(duì)數(shù)據(jù)流的操作。

以大文件拷貝程序?yàn)槔?，我們可以為?shù)據(jù)源創(chuàng)建一個(gè)只讀數(shù)據(jù)流，示例如下：

復(fù)制代碼代碼如下:

 var rs = fs.createReadStream(pathname);

 rs.on('data', function (chunk) {

     doSomething(chunk) ; // 具體細(xì)節(jié)自己任意發(fā)揮

 });

 rs.on('end', function () {

     cleanUp() ;

 }) ;

代碼中data事件會(huì)源源不斷地被觸發(fā)，不管doSomething函數(shù)是否處理得過(guò)來(lái)。代碼可以繼續(xù)做如下改造，以解決這個(gè)問(wèn)題。

復(fù)制代碼代碼如下:

 var rs = fs.createReadStream(src) ;

 rs.on('data', function (chunk) {

     rs.pause() ;

     doSomething(chunk, function () {

         rs.resume() ;

     }) ;

 }) ;

 rs.on('end', function () {

     cleanUp();

 })  ;

給doSomething函數(shù)加上了回調(diào)，因此我們可以在處理數(shù)據(jù)前暫停數(shù)據(jù)讀取，并在處理數(shù)據(jù)后繼續(xù)讀取數(shù)據(jù)。

此外，我們也可以為數(shù)據(jù)目標(biāo)創(chuàng)建一個(gè)只寫(xiě)數(shù)據(jù)流，如下：

復(fù)制代碼代碼如下:

 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換成了往只寫(xiě)數(shù)據(jù)流里寫(xiě)入數(shù)據(jù)后，以上代碼看起來(lái)就像是一個(gè)文件拷貝程序了。但是以上代碼存在上邊提到的問(wèn)題，如果寫(xiě)入速度跟不上讀取速度的話，只寫(xiě)數(shù)據(jù)流內(nèi)部的緩存會(huì)爆倉(cāng)。我們可以根據(jù).write方法的返回值來(lái)判斷傳入的數(shù)據(jù)是寫(xiě)入目標(biāo)了，還是臨時(shí)放在了緩存了，并根據(jù)drain事件來(lái)判斷什么時(shí)候只寫(xiě)數(shù)據(jù)流已經(jīng)將緩存中的數(shù)據(jù)寫(xiě)入目標(biāo)，可以傳入下一個(gè)待寫(xiě)數(shù)據(jù)了。因此代碼如下：

復(fù)制代碼代碼如下:

 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();

 }) ;

最終實(shí)現(xiàn)了數(shù)據(jù)從只讀數(shù)據(jù)流到只寫(xiě)數(shù)據(jù)流的搬運(yùn)，并包括了防爆倉(cāng)控制。因?yàn)檫@種使用場(chǎng)景很多，例如上邊的大文件拷貝程序，NodeJS直接提供了.pipe方法來(lái)做這件事情，其內(nèi)部實(shí)現(xiàn)方式與上邊的代碼類(lèi)似。

下面是一個(gè)更加完整的復(fù)制文件的過(guò)程：

復(fù)制代碼代碼如下:

var fs = require('fs'),

  path = require('path'),

  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 passedLength = 0;

var lastSize = 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 percent = Math.ceil((passedLength / totalSize) * 100);

  var size = Math.ceil(passedLength / 1000000);

  var diff = size - lastSize;

  lastSize = size;

  out.clearLine();

  out.cursorTo(0);

  out.write('已完成' + size + 'MB, ' + percent + '%, 速度：' + diff * 2 + 'MB/s');

  if (passedLength < totalSize) {

    setTimeout(show, 500);

  } else {

    var endTime = Date.now();

    console.log();

    console.log('共用時(shí)：' + (endTime - startTime) / 1000 + '秒。');

  }

}, 500);