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