node中的stream(前篇)

Stream在node中是一个无处不在的概念,但凡和IO沾边的程序都离不开stream,所以不弄懂stream是无法真正使用node的。在这份简短的笔记中,让我们来一起看看这个融入node血肉中的stream到底是什么,以及该如何正确使用stream吧。

Stream这个概念一点都不新鲜,早在unix早期就被用于命令行中了。在当时,计算机内部带宽非常有限,输入输出都只能按字节码顺序进行传输,这也就是最早的stream的概念。后来stream这个概念被发扬广大,几乎所有IO数据都按stream的形式,通过pipe(管道)进行传输,甚至连进程间通信都用到了stream和pipe的概念。

在node中,stream作为一个各种可读写的对象的抽象层,方便我们用统一的方法对各种对象进行操作。常见的http request response,标准输出流stdout和输入流stdin都可以用stream来进行处理。

node中的stream分为Readable(可读的)、Writable(可写)或是Duplex两者兼有的三大类,让我们依次了解吧。

stream.Readable

Readable stream是对所有可读数据源的抽象。node中的Readable stream有两个模式:flowing和paused。flowing指的是数据流不停歇地流入,一有数据就立刻传输,以便尽快将数据传输给出去;而paused模式则每次只传输一部分数据,读取了一部分数据后再传输这部分数据,按块传输数据直到传输结束。

前面说过,stream是通过pipe(管道)传输的。readable的pipe方法非常直观:

readable.pipe(destination[, options])

destination就是数据传输的目标。options接收一个end参数,该参数为true时,只要接收方明确停止数据传输,数据流就立即停止。

在node学习笔记中我们见过的Readable stream包括:

  • 客户端的http response
  • 服务器端的http request
  • fs模块中的read streams
  • tcp中的socket
  • process的stdin和stdout方法

下面让我们来看一个fs的例子。fs有一个createReadStream方法,可以按stream的方式读取文件。我们就用这个方法来读取命令行传入的指定路径的文件,然后用pipe(管道)传输给stdout:

var fs = require('fs');
fs.createReadStream(process.argv[2]).pipe(process.stdout);

另外一个简单的例子,stdin产生的流传输给stdout:

process.stdin.pipe(process.stdout);

如果你用node执行这个程序,就会看到每输入一行回车后,屏幕(stdout)就会重复这一行的内容。

Transform Stream

如果只是简单的将数据从一端传到另一端,那么流的也未免太简单了。事实上,stream是非常灵活的一种数据传输手段。现在我们就来看看transform stream是如何实现这种灵活性的。

许多时候,我们希望对数据先做一些处理,再将处理后的结果传输给接收方。这就是的transform stream的作用了,顾名思义,它的输出流是基于输入流变动得来的。它本身属于Deplux类型,可读可写。

我们来看一个简单例子,将前面例子的stdin输入流转换为大写后再传输给输出流stdout。

也许你会想到process.stdin.toUpperCase().pipe.(process.stdout),但这样行不通的。前面讲过,process.stdin的流是一个Readable stream,是不能被改变的。所以我们需要将readable stream转换为Deplux stream,再将内容改成大写,最后输出给stdout。

这个转换流的功能有些复杂,我们这里借助through2模块来做演示。

var through = require('through2');

使用through我们就可以创建一个Deplux流了:

var stream = through(transformFunction, flushFunction)

这里的transformFunction和flushFunction都是callback函数。transformFunction用来对出入流做处理,该函数的签名为:

transform._transform(chunk, encoding, callback)

chunk就是输入流的数据块,encoding是输入流的编码,callback在数据处理完成后调用。每个chunk处理完成后用push方法来写入,并转换为deplux stream进行输出。

flushFunction将在所有输入流传输结束后调用,用来结束transform stream。

有了这个模块的帮助,程序就可以写成:

var through = require('through2');
var stream = through(write, end);

function write(buffer, encoding, next) {
    this.push(buffer.toString().toUpperCase());
    next();
}

function end(done) {
    done();
}
process.stdin.pipe(stream).pipe(process.stdout);

用write函数将输入流逐块转换为大写,并用push转换为deplux stream。end则在输入流结束后发出done来明确结束transform stream。

下面再介绍一个很有用的模块split,可以将输入流按指定的分隔符分开。比如前面的输入的例子,我们可以按换行符来分割输入流,然后再按行输出:

process.stdin.pipe(split()).pipe(process.stdot);

另外如果我们不想按块或者按行处理,而是想在一起处理所有的输入流怎么办呢?我们需要一个方法将输入的数据块块连接起来,有个很方便的模块concat-stream可以完成这个任务:

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

注意这个concat的callback的buf参数不再是一个stream了,不能再使用pipe传输了,所以这里我们console就可以看到完整的输入了。

Stream与http服务

在node学习笔记里,我们介绍过http服务。我们说过,利用node核心的http模块,可以很方便的创建http服务。http.createServer()的callback带有两个参数,一个代表用户请求的request,还有一个代表响应的response。其实request和response都可以处理stream。比如可以将某个流输出到response中:pipe(response)。也可以将用户请求中附带的数据按流输入,比如request.pipe()。

这里我们来看一个例子,将request数据按stream输入,转换为大写后再通过response返回给客户:

var http = require('http');
var through = require('through2');

var server = http.createServer(function (req, res) {
  req.pipe(tr).pipe(res);
});

server.listen(process.argv[2]);

var tr = through(function (buf, _, next) {
  var buffer = buf.toString().toUpperCase();
  this.push(buffer);
  next();
});

内容前面都讲过,应该不陌生了。处理stream也很直观,只要按req.pipe(tr).pipe(res)就可以按前面提过的方法处理stream了。这么做的好处是无需等待所有数据传输完毕就可以开始处理和传输,能够尽快给客户响应。

最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 206,723评论 6 481
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 88,485评论 2 382
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 152,998评论 0 344
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 55,323评论 1 279
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 64,355评论 5 374
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 49,079评论 1 285
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 38,389评论 3 400
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 37,019评论 0 259
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 43,519评论 1 300
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,971评论 2 325
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 38,100评论 1 333
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 33,738评论 4 324
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 39,293评论 3 307
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 30,289评论 0 19
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,517评论 1 262
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 45,547评论 2 354
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 42,834评论 2 345

推荐阅读更多精彩内容

  • stream 流是一个抽象接口,在 Node 里被不同的对象实现。例如 request to an HTTP se...
    明明三省阅读 3,392评论 1 10
  • 流是Node中最重要的组件和模式之一。在社区里有一句格言说:让一切事务流动起来。这已经足够来描述在Node中流...
    宫若石阅读 543评论 0 0
  • Spring Cloud为开发人员提供了快速构建分布式系统中一些常见模式的工具(例如配置管理,服务发现,断路器,智...
    卡卡罗2017阅读 134,600评论 18 139
  • https://nodejs.org/api/documentation.html 工具模块 Assert 测试 ...
    KeKeMars阅读 6,305评论 0 6
  • Node.js是目前非常火热的技术,但是它的诞生经历却很奇特。 众所周知,在Netscape设计出JavaScri...
    w_zhuan阅读 3,609评论 2 41