Nodejs Stream 数据流使用手册

这篇文章主要介绍了数据流使用手册的相关资料，感兴趣的小伙伴一起学习吧、介绍本文介绍了使用开程序的基本方法。<codemizar">"WeshouldwaysoflikeinsegmentwhenitbecomesnecessarytomassageinisthewayofDougMcIlroy.October11,1964</code>最早接触Stream是早期的unix开始的数十年的实践证明Stream思想可以很简单的开发出一些庞大的系统Stream是过实现的在node中为内置的stream模块多心模块和三方模块都使用到unix一样主的操作也是pipe()，使用者可以使用反压力机制来控制读和写的平衡。可为开发者提供可以重复使用统一的接口，通过抽象的接口来控制之的写平衡。、为什么使用node中I/O是步的此对磁盘和网络的读写需要通过回调函数来读取数据是一个文件下载服务器的简单代码：<codeclass="hljsvarfsvarserver=http.createServer(function(req,{+function(err,{res.end(data);});});server.listen(8000);</code>这些代码可以实现需要的功能服务在发送文件数据之前需要缓存整个文件数据到内存，如"data.txt"件很大且并发量很大的话，会浪费很多内存。因为用户需要等到整个文件缓存到内存才能接受的文件数据，这样导致用户体验相当不好。不过还两个参数都是，这样我们可以用代替fs.readFile():<codeclass="hljsvarfsvarserver=http.createServer(function(req,{var=+'/data.txt');stream.pipe(res);});server.listen(8000);</code>.pipe()方法监听的data'和'end'事件样文就不需要缓存整个文件，当客户端连接完成之后马上可以发送一个数据块到客户端。使.pipe()另一个好处是可以解决当客户端延迟非常大时导致的读写不平衡问题果压缩文件再发送以使用三方模块实现：<codeclass="hljs

varfsvaroppressor=varserver=http.createServer(function(req,{var=+'/data.txt');});server.listen(8000);</code>这样文件就会对支持和的览器进行压缩。oppressor模会理所有的content-encoding。使发程序变得简单。、基础概念有五种基本的transform,duplex,and”classic”.3-1、pipe所有类型的收使用来创建一个输入输出对，接收一个可读流src并其数据输出到可写流，如下：<codeclass="hljsperl">src.pipe(dst)</code>.pipe()方法为返回流这就可以接连用多.pipe()，下：<codeclass="hljsperl">a.pipe(b).pipe(c).pipe()</code>功能与下面的代码相同：<codeclass="hljsperl">a.pipe(b);c);d);</code>3-2、readable通过调用Readable的法可以把Readablestreams的数据写入一个Writable,Transform,或Duplexstream。<codeclass="hljs)</code>创readablestream这里我们创建一个stream!<codeclass="hljsperl">varReadable=require('stream').Readable;varrs=newReadable;rs.push('beep');rs.push('boop\n');rs.push(null);rs.pipe(process.stdout);noderead0.jsbeep</code>null)通知数据接收者数据已经发送完注意到我们在将所有数据内容压入可读流之前并没有调用rs.pipe(process.stdout);但是我们压入的所有数据内容还是完全的输出了，这是因为可读流在接收者没有读取数据之前，会缓存所有压入的数据。但是在很多情况下,更好的方法是只有数接收着请求数据的时候，才压入数据到可读流而不是缓存整个数据。下面我们重写一函数：<codeclass="hljs=varrs=Readable();

varc=rs._read=(){rs.push(String.fromCharCode(c++));if'z'.charCodeAt(0))};<codeclass="hljsbash">$noderead1.js上面的代码通过重写_read()方法实现了只有在数据受者请求数据才向可读流中压入数据方也可以接收一个参表示数据请求着请求的数据大小是读流可以根据需要忽略这个参数。注意我们也可以用util.inherits()继可读流。为了说明只有在数据接受者请求数据时方法才被调用，我们在向可读流压入数据时做一个延时，如：<codeclass="hljs=varrs=Readable();varc=97-rs._read=(){if'z'.charCodeAt(0))returnsetTimeout(function(){rs.push(String.fromCharCode(++c));},};rs.pipe(process.stdout);process.on('exit',function(){console.error('\n_read()called+97)'times');});process.stdout.on('error',process.exit);</code>用下面的命令运行程序我们发_方法只调用了：<codeclass="hljsbash">$noderead2.js|abcde_read()使用计时器的原因是系统需时间来发信号来程序关闭管道。使用process.stdout.on('error',是了处理系统因为命关管道而发送SIGPIPE信，因为这样会导致process.stdout触事件。如果想创建一个的可以压入任意形式数据的可读流，只要在创建流的时候设置参数为即可，例如：objectMode:true})。读数大部分情况下我们只要简单的使用pipe方将读流的数据重定向到另外形式的流，但是在某些情况下也许直接从可读流中读取数据更有用。如下：<codeclass="hljs(){varbuf=console.dir(buf);});(echo1;1;ghi)nodeconsume0.js 

0a=""62=""63="">0a=""65=""66="">0a=""68=""69="">null</buffer></buffer></buffer></code>当可读流中有数据可读取时流触发'readable'事件这就可以调用.方法来读取相关数据可读流中没有数可读取时.read()会返回null这就可以结束.的调用，等下一次readable'事的触发下面是一个使从标准输入每次读取3个字节的例子：<codeclass="hljsfunction(){varbuf=process.stdin.read(3);console.dir(buf);如下运行程序发现，输出结果并不完!<codebash">$(echoabc;sleepechosleep1;echo|consume1.js 61=""62=""63="">0a=""65="">0a=""66=""这是应为额外的数据数据留在流的内部缓冲区里了们需要通知流我们要读取更多的数据可达到这目的。<codeclass="hljsfunction(){varbuf=process.stdin.read(3);console.dir(buf);process.stdin.read(0);这次运行结果如下：<codeabc;sleepdef;1;|nodeconsume2.js 0a=""65="">0a=""69=""></buffer></buffer></code>我们可以使用将据重新押回流数据队列的头部样以接续读取押回数据。如下面的代码，会按行输出标准输入的内容：<codeclass="hljs=0;function{varbuf=if(!buf)return;for(;offset<buf.length;{if===0x0a){offset).toString());=buf.slice(offset1);=0;return;

}}});tail-n/usr/share/dict/american-englishhead-n10|node'heartiest''heartland''heartland\'s''heartless'当然，有很多模块可以实现这个功能，如split。3-3、streamswritable只以作为pipe()函数的目的参数。如下代码：<codeclass="hljsperl">src.pipe();</code>创writablestream重写._write(chunk,enc,方就可以接受一个stream的据<codeclass="hljsphp">varWritablevarwsws._write=function{};(echobeep;sleepecho|write0.js 0a=""65=""70="">0a=""70=""></buffer></buffer></code>第一个参数chunk是据输入者写入的数据二个参数是数据的编码格式三个参数next(err)过回调函数通知数据写入者可以写入更多的时间。如果readable入的是字符串，那么字符串会默认转换为，如在创流的时候设置Writable({decodeStrings:false})参数，那么不会做转换。如果readablestream写的数据时对象，那么需要这样创建<codeclass="hljscss">Writable({写据到stream调用writablestream的.方即可完成数据写入。<codeclass="hljsvala">process.stdout.write('beepboop\n');</code>调用end()方法通知writable数已经写入完成。<codeclass="hljsfsvarws');

setTimeout(function(){},nodecatbeepboop</code>如果需要设置的冲区的大小那么在创建流的时候，需要设置opts.highWaterMark，样如果缓冲区里的数据超过opts.highWaterMark，方会返回false。当缓冲区可写时候writable会发'drain'事件。3-4、streamsClassic比老的接口了，最早出现在版中，但是了解一下其运行原理还是十分有好处的当个流被注册"事件的回到函数那么流就会工作在版本模式下会使用老的API。readableClassicreadable件就是一个事件触发器，如果Classicreadable有据可读取，那么其触发"data"事件，等到数据读取完毕时，会触发"事。.pipe()方通过检查的确定流是否有数据可读。下面是一个使用Classicreadablestreams打A-J字的例子：<codeclass="hljsrequire('stream');var=Stream;=varc=variv=setInterval(function(){if75){clearInterval(iv);}stream.emit('data',},nodeclassic0.js如果要从readable中取数据"data"和end"两个事件的回调函数即可，代码如下：<codeclass="hljs{console.log(buf);});(){console.log('__END__');});(echobeep;sleepecho|classic1.js 0a=""65=""70="">0a=""70="">

__END__</buffer></buffer></code>需要注意的是如果你使用这种方式读取数据么失去使用新接口带来的好处如你在往一个延迟常大的流写据时，需要注意读取数据和写数据的平衡问题，否则会导致大量数据缓存在内存中，导致浪费大量内存。一般这时候强烈建议使用流.pipe()方法，这样就不用自己监听”end”事件了，也不用担心读写不平衡的问题了。当然你也可以用替自己监听”data”和””事件如下面的代码：<codeclass="hljsphp">varthrough=process.stdin.pipe(through(write,end));functionwrite(buf){console.log(buf);}f