這篇文章主要講解了“分析Nodejs的Stream模塊”，文中的講解內(nèi)容簡單清晰，易于學(xué)習(xí)與理解，下面請大家跟著小編的思路慢慢深入，一起來研究和學(xué)習(xí)“分析Nodejs的Stream模塊”吧！
一，開篇分析
流是一個(gè)抽象接口，被
Node 中的很多對象所實(shí)現(xiàn)。比如對一個(gè) HTTP 服務(wù)器的請求是一個(gè)流，stdout 也是一個(gè)流。流是可讀，可寫或兼具兩者的。
最早接觸Stream是從早期的unix開始的， 數(shù)十年的實(shí)踐證明Stream 思想可以很簡單的開發(fā)出一些龐大的系統(tǒng)。
在unix里，Stream是通過 "|" 實(shí)現(xiàn)的。在node中，作為內(nèi)置的stream模塊，很多核心模塊和三方模塊都使用到。
和unix一樣，node stream主要的操作也是.pipe()，使用者可以使用反壓力機(jī)制來控制讀和寫的平衡。
Stream 可以為開發(fā)者提供可以重復(fù)使用統(tǒng)一的接口，通過抽象的Stream接口來控制Stream之間的讀寫平衡。
一個(gè)TCP連接既是可讀流，又是可寫流，而Http連接則不同，一個(gè)http request對象是可讀流，而http response對象則是可寫流。
流的傳輸過程默認(rèn)是以buffer的形式傳輸?shù)?，除非你給他設(shè)置其他編碼形式,以下是一個(gè)例子：
1.
2.
3.
var http = require('http') ;
4.
var server = http.createServer(function(req,res){
5.
res.writeHeader(200, {'Content-Type': 'text/plain'}) ;
6.
res.end("Hello,大熊!") ;
7.
}) ;
8.
server.listen(8888) ;
9.
console.log("http server running on port 8888 ...") ;
運(yùn)行后會有亂碼出現(xiàn)，原因就是沒有設(shè)置指定的字符集，比如：“utf-8” 。
修改一下就好：
var http = require('http') ;
1.
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 ...") ;
運(yùn)行結(jié)果：
為什么使用Stream
node中的I/O是異步的，因此對磁盤和網(wǎng)絡(luò)的讀寫需要通過回調(diào)函數(shù)來讀取數(shù)據(jù)，下面是一個(gè)文件下載例子
上代碼：
1.
2.
var http = require('http') ;
3.
var fs = require('fs') ;
4.
var server = http.createServer(function (req, res) {
5.
fs.readFile(__dirname + '/data.txt', function (err, data) {
6.
res.end(data);
7.
}) ;
8.
}) ;
9.
server.listen(8888) ;
代碼可以實(shí)現(xiàn)需要的功能，但是服務(wù)在發(fā)送文件數(shù)據(jù)之前需要緩存整個(gè)文件數(shù)據(jù)到內(nèi)存，如果"data.txt"文件很
大并且并發(fā)量很大的話，會浪費(fèi)很多內(nèi)存。因?yàn)橛脩粜枰鹊秸麄€(gè)文件緩存到內(nèi)存才能接受的文件數(shù)據(jù)，這樣導(dǎo)致
用戶體驗(yàn)相當(dāng)不好。不過還好(req,res)兩個(gè)參數(shù)都是Stream，這樣我們可以用fs.createReadStream()代替fs.readFile()。如下：
var http = require('http') ;
1.
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()方法監(jiān)聽fs.createReadStream()的'data' 和'end'事件，這樣"data.txt"文件就不需要緩存整
個(gè)文件，當(dāng)客戶端連接完成之后馬上可以發(fā)送一個(gè)數(shù)據(jù)塊到客戶端。使用.pipe()另一個(gè)好處是可以解決當(dāng)客戶
端延遲非常大時(shí)導(dǎo)致的讀寫不平衡問題。
有五種基本的Stream：readable，writable，transform，duplex，and "classic” 。(具體使用請自己查閱api)
二，實(shí)例引入
當(dāng)內(nèi)存中無法一次裝下需要處理的數(shù)據(jù)時(shí)，或者一邊讀取一邊處理更加高效時(shí)，我們就需要用到數(shù)據(jù)流。NodeJS中通過各種Stream來提供對數(shù)據(jù)流的操作。
以大文件拷貝程序?yàn)槔?，我們可以為?shù)據(jù)源創(chuàng)建一個(gè)只讀數(shù)據(jù)流，示例如下：
var rs = fs.createReadStream(pathname);
1.
2.
rs.on('data', function (chunk) {
3.
doSomething(chunk) ; // 具體細(xì)節(jié)自己任意發(fā)揮
4.
});
5.
rs.on('end', function () {
6.
cleanUp() ;
7.
}) ;
代碼中data事件會源源不斷地被觸發(fā)，不管doSomething函數(shù)是否處理得過來。代碼可以繼續(xù)做如下改造，以解決這個(gè)問題。
1.
2.
3.
var rs = fs.createReadStream(src) ;
4.
rs.on('data', function (chunk) {
5.
rs.pause() ;
6.
doSomething(chunk, function () {
7.
rs.resume() ;
8.
}) ;
9.
}) ;
10.
rs.on('end', function () {
11.
cleanUp();
12.
}) ;
給doSomething函數(shù)加上了回調(diào)，因此我們可以在處理數(shù)據(jù)前暫停數(shù)據(jù)讀取，并在處理數(shù)據(jù)后繼續(xù)讀取數(shù)據(jù)。
此外，我們也可以為數(shù)據(jù)目標(biāo)創(chuàng)建一個(gè)只寫數(shù)據(jù)流，如下：
var rs = fs.createReadStream(src) ;
1.
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換成了往只寫數(shù)據(jù)流里寫入數(shù)據(jù)后，以上代碼看起來就像是一個(gè)文件拷貝程序了。但是以上代碼存在上邊提到的問題，如果寫入速度跟不上讀取速度的話，只寫數(shù)據(jù)流內(nèi)部的緩存會爆倉。我們可以根據(jù).write方法的返回值來判斷傳入的數(shù)據(jù)是寫入目標(biāo)了，還是臨時(shí)放在了緩存了，并根據(jù)drain事件來判斷什么時(shí)候只寫數(shù)據(jù)流已經(jīng)將緩存中的數(shù)據(jù)寫入目標(biāo)，可以傳入下一個(gè)待寫數(shù)據(jù)了。因此代碼如下：
var rs = fs.createReadStream(src) ;
1.
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.
}) ;
最終實(shí)現(xiàn)了數(shù)據(jù)從只讀數(shù)據(jù)流到只寫數(shù)據(jù)流的搬運(yùn)，并包括了防爆倉控制。因?yàn)檫@種使用場景很多，例如上邊的大文件拷貝程序，NodeJS直接提供了.pipe方法來做這件事情，其內(nèi)部實(shí)現(xiàn)方式與上邊的代碼類似。
下面是一個(gè)更加完整的復(fù)制文件的過程：
var fs = require('fs'),
1.
2.
path = require('path'),
3.
out = process.stdout;
4.
var filePath = '/bb/bigbear.mkv';
5.
var readStream = fs.createReadStream(filePath);
6.
var writeStream = fs.createWriteStream('file.mkv');
7.
var stat = fs.statSync(filePath);
8.
var totalSize = stat.size;
9.
var passedLength = 0;
10.
var lastSize = 0;
11.
var startTime = Date.now();
12.
readStream.on('data', function(chunk) {
13.
passedLength += chunk.length;
14.
if (writeStream.write(chunk) === false) {
15.
readStream.pause();
16.
}
17.
});
18.
readStream.on('end', function() {
19.
writeStream.end();
20.
});
21.
writeStream.on('drain', function() {
22.
readStream.resume();
23.
});
24.
setTimeout(function show() {
25.
var percent = Math.ceil((passedLength / totalSize) * 100);
26.
var size = Math.ceil(passedLength / 1000000);
27.
var diff = size - lastSize;
28.
lastSize = size;
29.
out.clearLine();
30.
out.cursorTo(0);
31.
out.write('已完成' + size + 'MB, ' + percent + '%, 速度：' + diff * 2 +
32. 'MB/s');
33.
if (passedLength < totalSize) {
34.
setTimeout(show, 500);
35.
} else {
36.
var endTime = Date.now();
37.
console.log();
38.
console.log('共用時(shí)：' + (endTime - startTime) / 1000 + '秒。');
39.
}
40.
}, 500);
可以把上面的代碼保存為
"copy.js" 試驗(yàn)一下我們添加了一個(gè)遞歸的 setTimeout (或者直接使用setInterval)來做一個(gè)旁觀者，
每500ms觀察一次完成進(jìn)度，并把已完成的大小、百分比和復(fù)制速度一并寫到控制臺上，當(dāng)復(fù)制完成時(shí)，計(jì)算總的耗費(fèi)時(shí)間。
感謝各位的閱讀，以上就是“分析Nodejs的Stream模塊”的內(nèi)容了，經(jīng)過本文的學(xué)習(xí)后，相信大家對分析Nodejs的Stream模塊這一問題有了更深刻的體會，具體使用情況還需要大家實(shí)踐驗(yàn)證。這里是創(chuàng)新互聯(lián)，小編將為大家推送更多相關(guān)知識點(diǎn)的文章，歡迎關(guān)注！
重慶分公司
重慶分公司
