nodejs流

--如果你想成为一个Node高手，那么流一定是武功秘籍中不可缺少的一个部分

详细讲解stream以及一些实例

https://github.com/substack/stream-handbook/blob/master/example/writable.js

1.为什么使用流

之前写过如下代码，有问题！
消耗大量的内存
为用户在接收到任何的内容之前首先需要等待程序将文件内容完全读入到内存中

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

解决之道
(req,res)参数都是流对象

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

在这里，.pipe()方法会自动帮助我们监听data和end事件。上面的这段代码不仅简洁，而且data.txt文件中每一小段数据都将源源不断的发送到客户端。
除此之外，使用.pipe()方法还有别的好处，比如说它可以自动控制后端压力，以便在客户端连接缓慢的时候node可以将尽可能少的缓存放到内存中。

2.想要将数据进行压缩？可以使用相应的流模块完成这项工作!

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

3.流模块基础

在node中，一共有五种类型的流readable,writable,transform,duplex以及"classic"

（1）pipe

无论哪一种流，都会使用.pipe()方法来实现输入和输出。
.pipe(dst)将会返回dst因此你可以链式调用多个流:

a.pipe(b).pipe(c).pipe(d)
上面的代码也可以等价为：
a.pipe(b);
b.pipe(c);
c.pipe(d);
(2)readable流

Readable流可以产出数据，你可以将这些数据传送到一个writable，transform或者duplex流中，只需要调用pipe()方法:

readableStream.pipe(dst)

创建一个readable流

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

在上面的代码中rs.push(null)的作用是告诉rs
输出数据应该结束了
需要注意的一点是我们在将数据输出到process.stdout之前已经将内容推送进readable流rs中，但是所有的数据依然是可写的。
这是因为在你使用.push()将数据推进一个readable流中时，一直要到另一个东西来消耗数据之前，数据都会存在一个缓存中。
解决之道
我们可以通过定义一个._read函数来实现按需推送数据:

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

运行结果

$ node read1.js
abcdefghijklmnopqrstuvwxyz

在这里我们将字母a到z推进了rs中，但是只有当数据消耗者出现时，数据才会真正实现推送。
为了说明只有在数据消耗者出现时，_read函数才会被调用，我们可以将上面的代码简单的修改一下：

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

运行上面的代码我们可以发现如果我们只请求5比特的数据，那么_read
只会运行5次：

$ node read2.js | head -c5
abcde
_read() called 5 times

另外,process.stdout.on('error',fn)处理器也很重要，因为当head
不再关心我们的程序输出时，操作系统将会向我们的进程发送一个SIGPIPE信号，此时process.stdout将会捕获到一个EPIPE错误。
如果你创建了一个readable流，并且想要将任何的值推送到其中的话，确保你在创建流的时候指定了objectMode参数,Readable({ objectMode: true })。

消耗一个readable流

大部分时候，将一个readable流直接pipe到另一种类型的流或者使用through或者concat-stream创建的流中，是一件很容易的事情。但是有时我们也会需要直接来消耗一个readable流。

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 61 62 63 0a>
<Buffer 64 65 66 0a>
<Buffer 67 68 69 0a>
null

当数据可用时，readable事件将会被触发，此时你可以调用.read()方法来从缓存中获取这些数据。
当流结束时，.read()将返回null，因为此时已经没有更多的字节可以供我们获取了。
你也可以告诉.read()方法来返回n个字节的数据。虽然所有核心对象中的流都支持这种方式，但是对于对象流来说这种方法并不可用。
下面是一个例子，在这里我们制定每次读取3个字节的数据：

process.stdin.on('readable', function () {
    var buf = process.stdin.read(3);
    console.dir(buf);
});

运行上面的例子，我们将获取到不完整的数据:

$ (echo abc; sleep 1; echo def; sleep 1; echo ghi) | node consume1.js 
<Buffer 61 62 63>
<Buffer 0a 64 65>
<Buffer 66 0a 67>

这是因为多余的数据都留在了内部的缓存中，因此这个时候我们需要告诉node我们还对剩下的数据感兴趣，我们可以使用.read(0)来完成这件事：

process.stdin.on('readable', function () {
    var buf = process.stdin.read(3);
    console.dir(buf);
    process.stdin.read(0);
});

到现在为止我们的代码和我们所期望的一样了！

$ (echo abc; sleep 1; echo def; sleep 1; echo ghi) | node consume2.js 
<Buffer 61 62 63>
<Buffer 0a 64 65>
<Buffer 66 0a 67>
<Buffer 68 69 0a>

4.writable流

一个writable流指的是只能流进不能流出的流

src.pipe(writableStream)
(1)创建一个writable流
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 62 65 65 70 0a>
<Buffer 62 6f 6f 70 0a>

第一个参数，chunk代表写进来的数据。

第二个参数enc代表编码的字符串，但是只有在opts.decodeString为false的时候你才可以写一个字符串。

第三个参数，next(err)是一个回调函数，使用这个回调函数你可以告诉数据消耗者可以写更多的数据。你可以有选择性的传递一个错误对象error，这时会在流实体上触发一个emit事件。
在从一个readable流向一个writable流传数据的过程中，数据会自动被转换为Buffer对象，除非你在创建writable流的时候制定了decodeStrings
参数为false,Writable({decodeStrings: false})。

向一个writable流中写东西

如果你需要向一个writable流中写东西，只需要调用.write(data)
即可。

process.stdout.write('beep boop\n');

为了告诉一个writable流你已经写完毕了，只需要调用.end()
方法。你也可以使用.end(data)在结束前再写一些数据

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

如果你想要等待缓存情况，可以监听drain事件。

5.transform流

你可以将transform流想象成一个流的中间部分，它可以读也可写，但是并不保存数据，它只负责处理流经它的数据。

6.duplex流

Duplex流是一个可读也可写的流，就好像一个电话，可以接收也可以发送语音。一个rpc交换是一个duplex流的最好的例子。如果你看到过下面这样的代码：

a.pipe(b).pipe(a)

7.classic流

Classic流是一个古老的接口，最早出现在node 0.4中。虽然现在不怎么用，但是我们最好还是来了解一下它的工作原理。
无论何时，只要一个流对象注册了一个data监听器，它就会自动的切换到classic模式，并且根据旧API的方式运行。

1.classic readable流

Classic readable流只是一个事件发射器，当有数据消耗者出现时发射emit事件，当输出数据完毕时发射end事件。
我们可以同构检查stream.readable来检查一个classic流对象是否可读。
下面是一个简单的readable流对象的例子，程序的运行结果将会输出A
到J：

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

为了从一个classic readable流中读取数据，你可以注册data和end
监听器。下面是一个使用旧readable流方式从process.stdin中读取数据的例子:

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 62 65 65 70 0a>
<Buffer 62 6f 6f 70 0a>
__END__

需要注意的一点是当你在一个流对象上注册了一个data
监听器，你就将这个流放在了兼容模式下，此时你不能使用两个stream2的api。
如果你自己创建流对象，永远不要绑定data和end监听器。如果你需要和旧版本的流兼容，最好使用第三方库来实现.pipe()方法。
例如，你可以使用through模块来避免显式的使用data和end监听器:

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 62 65 65 70 0a>
<Buffer 62 6f 6f 70 0a>
__END__

你也可以使用concat-stream模块来将整个流的内容缓存起来:

var concat = require('concat-stream');
process.stdin.pipe(concat(function (body) {
    console.log(JSON.parse(body));
}));

程序运行结果如下所示:

$ echo '{"beep":"boop"}' | node concat.js 
{ beep: 'boop' }

Classic readable流拥有.pause()和.resume()逻辑来暂停一个流，但是这都是可选的。如果你想要使用.pause()和.resume()方法，你应该使用through模块来帮助你处理缓存。
classic writable流

Classic writable流非常简单。其中只定义了.write(buf)，.end(buf)
，以及.desctory()方法。其中.end(buf)的参数buf是可选参数，但是一般来说node程序员还是喜欢使用.end(buf)这种写法。

demo1

var Stream = require('stream');
var s = new Stream;
s.writable = true;
var bytes = 0;
s.write = function (buf) {
bytes += buf.length;
};
s.end = function (buf) {
if (arguments.length) s.write(buf);
s.writable = false;
console.log(bytes + ' bytes written');
};

s.destroy = function () {
s.writable = false;
};

var fs = require('fs');
fs.createReadStream('/etc/passwd').pipe(s);
最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 204,732评论 6赞 478
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 87,496评论 2赞 381
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 151,264评论 0赞 338
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 54,807评论 1赞 277
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 63,806评论 5赞 368
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 48,675评论 1赞 281
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 38,029评论 3赞 399
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 36,683评论 0赞 258
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 41,704评论 1赞 299
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,666评论 2赞 321
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,773评论 1赞 332
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 33,413评论 4赞 321
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 39,016评论 3赞 307
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,978评论 0赞 19
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,204评论 1赞 260
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 45,083评论 2赞 350
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 42,503评论 2赞 343

推荐阅读更多精彩内容