Nodejs Stream 数据流使用手册

1、介绍

本文介绍了使用 node.js streams 开发程序的基本方法。

<code class="hljs mizar">"We should have some ways of connecting programs like garden hose--screw in
another segment when it becomes necessary to massage data in
another way. This is the way of IO also."
Doug McIlroy. October 11, 1964</code>

最早接触Stream是从早期的unix开始的数十年的实践证明Stream 思想可以很简单的开发出一些庞大的系统。在unix里,Stream是通过 |实现的;在node中,作为内置的stream模块,很多核心模块和三方模块都使用到。和unix一样,node Stream主要的操作也是.pipe(),使用者可以使用反压力机制来控制读和写的平衡。

Stream 可以为开发者提供可以重复使用统一的接口,通过抽象的Stream接口来控制Stream之间的读写平衡。

2、为什么使用Stream

node中的I/O是异步的,因此对磁盘和网络的读写需要通过回调函数来读取数据,下面是一个文件下载服务器的简单代码:

<code class="hljs javascript">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(8000);</code>

这些代码可以实现需要的功能,但是服务在发送文件数据之前需要缓存整个文件数据到内存,如果"data.txt"文件很大且并发量很大的话,会浪费很多内存。因为用户需要等到整个文件缓存到内存才能接受的文件数据,这样导致用户体验相当不好。不过还好(req, res)两个参数都是Stream,这样我们可以用fs.createReadStream()代替fs.readFile():

<code class="hljs javascript">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(8000);</code>

.pipe()方法监听fs.createReadStream()的'data' 和'end'事件,这样"data.txt"文件就不需要缓存整个文件,当客户端连接完成之后马上可以发送一个数据块到客户端。使用.pipe()另一个好处是可以解决当客户端延迟非常大时导致的读写不平衡问题。如果想压缩文件再发送,可以使用三方模块实现:

<code class="hljs javascript">var http = require('http');
var fs = require('fs');
var oppressor = require('oppressor');
var server = http.createServer(function (req, res) {
var stream = fs.createReadStream(__dirname + '/data.txt');
stream.pipe(oppressor(req)).pipe(res);
});
server.listen(8000);</code>

这样文件就会对支持gzip和deflate的浏览器进行压缩。oppressor 模块会处理所有的content-encoding。

Stream使开发程序变得简单。

3、基础概念

有五种基本的Stream: readable, writable, transform, duplex, and”classic”.

3-1、pipe

所有类型的Stream收是使用 .pipe() 来创建一个输入输出对,接收一个可读流src并将其数据输出到可写流dst,如下:

<code class="hljs perl">src.pipe(dst)</code>

.pipe( dst )方法为返回dst流,这样就可以接连使用多个.pipe(),如下:

<code class="hljs perl">a.pipe( b ).pipe( c ).pipe( d )</code>

功能与下面的代码相同:

<code class="hljs perl">a.pipe( b );
b.pipe( c );
c.pipe( d );</code>

3-2、readable streams

通过调用Readable streams的 .pipe()方法可以把Readable streams的数据写入一个Writable , Transform, 或者Duplex stream。

<code class="hljs perl">readableStream.pipe( dst )</code>

1>创建 readable stream

这里我们创建一个readable stream!

<code class="hljs perl">var Readable = require('stream').Readable;
var rs = new Readable;
rs.push('beep ');
rs.push('boop\n');
rs.push(null);
rs.pipe(process.stdout);
$ node read0.js
beep boop
</code>

rs.push( null ) 通知数据接收者数据已经发送完毕.

注意到我们在将所有数据内容压入可读流之前并没有调用rs.pipe(process.stdout);,但是我们压入的所有数据内容还是完全的输出了,这是因为可读流在接收者没有读取数据之前,会缓存所有压入的数据。但是在很多情况下, 更好的方法是只有数据接收着请求数据的时候,才压入数据到可读流而不是缓存整个数据。下面我们重写 一下._read()函数:

<code class="hljs javascript">var Readable = require('stream').Readable;
var rs = Readable();
var c = 97;
rs._read = function () {
rs.push(String.fromCharCode(c++));
if (c > 'z'.charCodeAt(0)) rs.push(null);
};
rs.pipe(process.stdout);</code>
<code class="hljs bash">$ node read1.js
abcdefghijklmnopqrstuvwxyz</code>

上面的代码通过重写_read()方法实现了只有在数据接受者请求数据才向可读流中压入数据。_read()方法也可以接收一个size参数表示数据请求着请求的数据大小,但是可读流可以根据需要忽略这个参数。

注意我们也可以用util.inherits()继承可读流。为了说明只有在数据接受者请求数据时_read()方法才被调用,我们在向可读流压入数据时做一个延时,如下:

<code class="hljs javascript">var Readable = require('stream').Readable;
var rs = Readable();
var c = 97 - 1;
rs._read = function () {
if (c >= 'z'.charCodeAt(0)) return rs.push(null);
setTimeout(function () {
rs.push(String.fromCharCode(++c));
}, 100);
};
rs.pipe(process.stdout);
process.on('exit', function () {
console.error('\n_read() called ' + (c - 97) + ' times');
});
process.stdout.on('error', process.exit);</code>

用下面的命令运行程序我们发现_read()方法只调用了5次:

<code class="hljs bash">$ node read2.js | head -c5
abcde
_read() called 5 times</code>

使用计时器的原因是系统需要时间来发送信号来通知程序关闭管道。使用process.stdout.on('error', fn) 是为了处理系统因为header命令关闭管道而发送SIGPIPE信号,因为这样会导致process.stdout触发EPIPE事件。如果想创建一个的可以压入任意形式数据的可读流,只要在创建流的时候设置参数objectMode为true即可,例如:Readable({ objectMode: true })。

2>读取readable stream数据

大部分情况下我们只要简单的使用pipe方法将可读流的数据重定向到另外形式的流,但是在某些情况下也许直接从可读流中读取数据更有用。如下:

<code class="hljs php">process.stdin.on('readable', function () {
var buf = process.stdin.read();
console.dir(buf);
});
$ (echo abc; sleep 1; echo def; sleep 1; echo ghi) | node consume0.js
<buffer 0a="" 61="" 62="" 63="">
<buffer 0a="" 64="" 65="" 66="">
<buffer 0a="" 67="" 68="" 69="">
null</buffer></buffer></buffer></code>

当可读流中有数据可读取时,流会触发'readable' 事件,这样就可以调用.read()方法来读取相关数据,当可读流中没有数据可读取时,.read() 会返回null,这样就可以结束.read() 的调用, 等待下一次'readable' 事件的触发。下面是一个使用.read(n)从标准输入每次读取3个字节的例子:

<code class="hljs javascript">process.stdin.on('readable', function () {
var buf = process.stdin.read(3);
console.dir(buf);
});</code>

如下运行程序发现,输出结果并不完全!

<code class="hljs bash">$ (echo abc; sleep 1; echo def; sleep 1; echo ghi) | node consume1.js
<buffer 61="" 62="" 63="">
<buffer 0a="" 64="" 65="">
<buffer 0a="" 66="" 67=""></buffer></buffer></buffer></code>

这是应为额外的数据数据留在流的内部缓冲区里了,而我们需要通知流我们要读取更多的数据.read(0)可以达到这个目的。

<code class="hljs javascript">process.stdin.on('readable', function () {
var buf = process.stdin.read(3);
console.dir(buf);
process.stdin.read(0);
});</code>

这次运行结果如下:

<code class="hljs xml">$ (echo abc; sleep 1; echo def; sleep 1; echo ghi) | node consume2.js
<buffer 0a="" 64="" 65="">
<buffer 0a="" 68="" 69=""></buffer></buffer></code>

我们可以使用 .unshift() 将数据重新押回流数据队列的头部,这样可以接续读取押回的数据。如下面的代码,会按行输出标准输入的内容:

<code class="hljs javascript">var offset = 0;
process.stdin.on('readable', function () {
var buf = process.stdin.read();
if (!buf) return;
for (; offset < buf.length; offset++) {
if (buf[offset] === 0x0a) {
console.dir(buf.slice(0, offset).toString());
buf = buf.slice(offset + 1);
offset = 0;
process.stdin.unshift(buf);
return;
}
}
process.stdin.unshift(buf);
});
$ tail -n +50000 /usr/share/dict/american-english | head -n10 | node lines.js
'hearties'
'heartiest'
'heartily'
'heartiness'
'heartiness\'s'
'heartland'
'heartland\'s'
'heartlands'
'heartless'
'heartlessly'</code>

当然,有很多模块可以实现这个功能,如:split 。

3-3、writable streams

writable streams只可以作为.pipe()函数的目的参数。如下代码:

<code class="hljs perl">src.pipe( writableStream );</code>

1>创建 writable stream

重写 ._write(chunk, enc, next) 方法就可以接受一个readable stream的数据。

<code class="hljs php">var Writable = require('stream').Writable;
var ws = Writable();
ws._write = function (chunk, enc, next) {
console.dir(chunk);
next();
};
process.stdin.pipe(ws);
$ (echo beep; sleep 1; echo boop) | node write0.js
<buffer 0a="" 62="" 65="" 70="">
<buffer 0a="" 62="" 6f="" 70=""></buffer></buffer></code>

第一个参数chunk是数据输入者写入的数据。第二个参数end是数据的编码格式。第三个参数next(err)通过回调函数通知数据写入者可以写入更多的时间。如果readable stream写入的是字符串,那么字符串会默认转换为Buffer,如果在创建流的时候设置Writable({ decodeStrings: false })参数,那么不会做转换。如果readable stream写入的数据时对象,那么需要这样创建writable stream

<code class="hljs css">Writable({ objectMode: true })</code>

2>写数据到 writable stream

调用writable stream的.write(data)方法即可完成数据写入。

<code class="hljs vala">process.stdout.write('beep boop\n');</code>

调用.end()方法通知writable stream 数据已经写入完成。

<code class="hljs javascript">var fs = require('fs');
var ws = fs.createWriteStream('message.txt');
ws.write('beep ');
setTimeout(function () {
ws.end('boop\n');
}, 1000);
$ node writing1.js
$ cat message.txt
beep boop</code>

如果需要设置writable stream的缓冲区的大小,那么在创建流的时候,需要设置opts.highWaterMark,这样如果缓冲区里的数据超过opts.highWaterMark,.write(data)方法会返回false。当缓冲区可写的时候,writable stream会触发'drain' 事件。

3-4、classic streams

Classic streams比较老的接口了,最早出现在node 0.4版本中,但是了解一下其运行原理还是十分有好
处的。当一个流被注册了"data" 事件的回到函数,那么流就会工作在老版本模式下,即会使用老的API。

1>classic readable streams

Classic readable streams事件就是一个事件触发器,如果Classic readable streams有数据可读取,那么其触发 "data" 事件,等到数据读取完毕时,会触发"end" 事件。.pipe() 方法通过检查stream.readable 的值确定流是否有数据可读。下面是一个使用Classic readable streams打印A-J字母的例子:

<code class="hljs javascript">var Stream = require('stream');
var stream = new Stream;
stream.readable = true;
var c = 64;
var iv = setInterval(function () {
if (++c >= 75) {
clearInterval(iv);
stream.emit('end');
}
else stream.emit('data', String.fromCharCode(c));
}, 100);
stream.pipe(process.stdout);
$ node classic0.js
ABCDEFGHIJ</code>

如果要从classic readable stream中读取数据,注册"data" 和"end"两个事件的回调函数即可,代码如下:

<code class="hljs php">process.stdin.on('data', function (buf) {
console.log(buf);
});
process.stdin.on('end', function () {
console.log('__END__');
});
$ (echo beep; sleep 1; echo boop) | node classic1.js
<buffer 0a="" 62="" 65="" 70="">
<buffer 0a="" 62="" 6f="" 70="">
__END__</buffer></buffer></code>

需要注意的是如果你使用这种方式读取数据,那么会失去使用新接口带来的好处。比如你在往一个 延迟非常大的流写数据时,需要注意读取数据和写数据的平衡问题,否则会导致大量数据缓存在内存中,导致浪费大量内存。一般这时候强烈建议使用流的.pipe()方法,这样就不用自己监听”data” 和”end”事件了,也不用担心读写不平衡的问题了。当然你也可以用 through代替自己监听”data” 和”end” 事件,如下面的代码:

<code class="hljs php">var through = require('through');
process.stdin.pipe(through(write, end));
function write (buf) {
console.log(buf);
}
function end () {
console.log('__END__');
}
$ (echo beep; sleep 1; echo boop) | node through.js
<buffer 0a="" 62="" 65="" 70="">
<buffer 0a="" 62="" 6f="" 70="">
__END__</buffer></buffer></code>

或者也可以使用concat-stream来缓存整个流的内容:

<code class="hljs oxygene">var concat = require('concat-stream');
process.stdin.pipe(concat(function (body) {
console.log(JSON.parse(body));
}));
$ echo '{"beep":"boop"}' | node concat.js
{ beep: 'boop' }</code>

当然如果你非要自己监听"data" 和"end"事件,那么你可以在写数据的流不可写的时候使用.pause()方法暂停Classic readable streams继续触发”data” 事件。等到写数据的流可写的时候再使用.resume() 方法通知流继续触发"data" 事件继续读取
数据。

2>classic writable streams

Classic writable streams 非常简单。只有 .write(buf), .end(buf)和.destroy()三个方法。.end(buf) 方法的buf参数是可选的,如果选择该参数,相当于stream.write(buf); stream.end() 这样的操作,需要注意的是当流的缓冲区写满即流不可写时.write(buf)方法会返回false,如果流再次可写时,流会触发drain事件。

4、transform

transform是一个对读入数据过滤然输出的流。

5、duplex

duplex stream是一个可读也可写的双向流,如下面的a就是一个duplex stream:

<code class="hljs livecodeserver">a.pipe(b).pipe(a)</code>

以上内容是小编给大家介绍的Nodejs Stream 数据流使用手册,希望对大家有所帮助!

(0)

相关推荐

  • 深入nodejs中流(stream)的理解

    nodejs的fs模块并没有提供一个copy的方法,但我们可以很容易的实现一个,比如: var source = fs.readFileSync('/path/to/source', {encoding: 'utf8'}); fs.writeFileSync('/path/to/dest', source); 这种方式是把文件内容全部读入内存,然后再写入文件,对于小型的文本文件,这没有多大问题,比如grunt-file-copy就是这样实现的.但是对于体积较大的二进制文件,比如音频.视频文件,动

  • Node.js中流(stream)的使用方法示例

    前言 本文主要给大家介绍了关于Node.js 流(stream)的使用方法,分享出来供大家参考学习,下面话不多说,来一起看看详细的介绍: 流是基于事件的API,用于管理和处理数据,而且有不错的效率.借助事件和非阻塞I/O库,流模块允许在其可用的时候动态处理,在其不需要的时候释放掉. 使用流的好处 举一个读取文件的例子: 使用fs.readFileSync同步读取一个文件,程序会被阻塞,所有的数据都会被读取到内存中. 换用fs.readFile读取文件,程序不会被阻塞,但是所有的数据依旧会被一次性

  • 详解Node.js串行化流程控制

    串行任务:需要一个接着一个坐的任务叫做串行任务. 可以使用回调的方式让几个异步任务按顺序执行,但如果任务过多,必须组织一下,否则过多的回调嵌套会把代码搞得很乱. 为了用串行化流程控制让几个异步任务按顺序执行,需要先把这些任务按预期的执行顺序放到一个数组中,这个数组将起到队列的作用:完成一个任务后按顺序从数组中取出下一个. 数组中的每个任务都是一个函数.任务完成后应该调用一个处理器函数,告诉它错误状态和结果. 为了演示如何实现串行化流程控制,我们准备做个小程序,让它从一个随机选择的RSS预定源中获

  • 用NODE.JS中的流编写工具是要注意的事项

    Node.js中的流十分强大,它对处理潜在的大文件提供了支持,也抽象了一些场景下的数据处理和传递.正因为它如此好用,所以在实战中我们常常基于它来编写一些工具 函数/库 ,但往往又由于自己对流的某些特性的疏忽,导致写出的 函数/库 在一些情况会达不到想要的效果,或者埋下一些隐藏的地雷.本文将会提供两个在编写基于流的工具时,私以为有些用的两个tips. 一,警惕EVENTEMITTER内存泄露 在一个可能被多次调用的函数中,如果需要给流添加事件监听器来执行某些操作.那么则需要警惕添加监听器而导致的内

  • Node.js中的缓冲与流模块详细介绍

    缓冲(buffer)模块 js起初就是为浏览器而设计的,所以能很好的处理unicode编码的字符串,但不能很好的处理二进制数据.这是Node.js的一个问题,因为Node.js旨在网络上发送和接收经常是以二进制格式传输的数据.比如: - 通过TCP连接发送和接收数据:  - 从图像或者压缩文件读取二进制数据:  - 从文件系统读写数据:  - 处理来自网络的二进制数据流 而Buffer模块为Node.js带来了一种存储原始数据的方法,于是可以再js的上下文中使用二进制数据.每当需要在Node.j

  • Nodejs Stream 数据流使用手册

    1.介绍 本文介绍了使用 node.js streams 开发程序的基本方法. <code class="hljs mizar">"We should have some ways of connecting programs like garden hose--screw in another segment when it becomes necessary to massage data in another way. This is the way of

  • 详解java8中的Stream数据流

    Stream是java8引入的一个重度使用lambda表达式的API.Stream使用一种类似用SQL语句从数据库查询数据的直观方式来提供一种对Java集合运算和表达的高阶抽象.直观意味着开发者在写代码时只需关注他们想要的结果是什么而无需关注实现结果的具体方式.这一章节中,我们将介绍为什么我们需要一种新的数据处理API.Collection和Stream的不同之处以及如何将StreamAPI应用到我们的编码中. 筛选重复的元素 Stream 接口支持 distinct 的方法, 它会返回一个元素

  • Node的stream数据流你了解吗

    目录 一.Node的数据流(stream) 1.处理缓存的方式 2.四种类型的流 3.zlib模块:用于文件的压缩与解压缩 (1)压缩方法:zlib.createGzip() (2)解压缩方法:zlib.createGunzip() 总结 一.Node的数据流(stream) 1.处理缓存的方式 (1)一次性全部读取,然后再进行处理.缺点是大文件处理非常耗时,优点是过程直观 (2)读取一块处理一块.优点是提高程序的性能 2.四种类型的流 Readable - 其是用于读操作 Writable -

  • 详解nodeJs文件系统(fs)与流(stream)

    一.简介 本文将介绍node.js文件系统(fs)和流(stream)的一些API已经参数使用情况. 二.目录 文件系统将介绍以下方法: 1.fs.readFile 2.fs.writeFile 3.fs.open 4.fs.read 5.fs.stat 6.fs.close 7.fs.mkdir 8.fs.rmdir 9.fs.readdir 10.fs.unlink stream流的四种类型readable,writable,duplex,transform以及stream对象的事件. 三.

  • Java8如何构建一个Stream示例详解

    Stream初体验 Stream是Java8中操作集合的一个重要特性,我们先来看看Java里面是怎么定义Stream的: "A sequence of elements supporting sequential and parallel aggregate operations." 我们来解读一下上面的那句话: 1.Stream是元素的集合,这点让Stream看起来用些类似Iterator: 2.可以支持顺序和并行的对原Stream进行汇聚的操作. Stream的创建方式有很多种,除

  • nodejs中使用archive压缩文件的实现代码

    前言 archive是一款在nodejs中可以实现跨平台打包的工具 可以将文件压缩为zip或rar格式 是一个比较好用的第三方模块 install npm install archiver --save archive github地址:https://github.com/archiverjs/node-archiver Quick Start // require modules var fs = require('fs'); var archiver = require('archiver

  • .net 通过URL推送POST数据具体实现

    由于到了一家新公司重新开始接触MVC和其他的一些东西.所以的重新拾起许多东西. 前一段时间让我写一个和第三方公司推送对接的方法.通过对方提供的URL把数据post推送出去. 我把url到了web.config里 复制代码 代码如下: <add key="urlStrings" value="urladdress"/> 在.CS文件里 复制代码 代码如下: private string postString = System.Configuration.C

  • IIS漏洞整理大全

    去年下半年关于IIS的漏洞层出不穷,鉴于目前IIS的广泛使用,觉得有必要把收集的资料加以总结一下. 1.介绍 这里介绍的方法主要通过端口80来完成操作,具有很大的威胁性,因为作为网络服务器80端口总要打开的.如果想方便一些,下载一些WWW.CGI扫描器来辅助检查. 而且要知道目标机器运行的是何种服务程序,你可以使用以下命令: telnet <目标机> 80 GET HEAD / HTTP/1.0 就可以返回一些域名和WEB服务程序版本,如果有些服务器把WEB服务运行在8080,81,8000,

  • Nodejs学习笔记之Stream模块

    一,开篇分析 流是一个抽象接口,被 Node 中的很多对象所实现.比如对一个 HTTP 服务器的请求是一个流,stdout 也是一个流.流是可读,可写或兼具两者的. 最早接触Stream是从早期的unix开始的, 数十年的实践证明Stream 思想可以很简单的开发出一些庞大的系统. 在unix里,Stream是通过 "|" 实现的.在node中,作为内置的stream模块,很多核心模块和三方模块都使用到. 和unix一样,node stream主要的操作也是.pipe(),使用者可以使

随机推荐