[Node.js] Node.js中的流

时间:2021-03-13 04:00:32

原文地址:http://www.moye.me/2015/03/29/streaming_in_node/

什么是流?

说到流,就涉及到一个*nix的概念:管道——在*nix中,流在Shell中被实现为可以通过 |(管道符) 进行桥接的数据,一个进程的输出(stdout)可被直接作为下一个进程的输入(stdin)。

在Node中,流(Stream)的概念与之类似,代表一种数据流可供桥接的能力。

pipe

[Node.js] Node.js中的流

流化的精髓在于 .pipe()方法。可供桥接的能力,在于数据流的两端(上游/下游 或称为 读/写流)以一个 .pipe()方法进行桥接。

伪代码的表现形式为:

//上游.pipe(下游)
Readable.pipe(Writable);

流的分类

这里并不打算讨论所谓的Node  v0.4 之前的“经典”流。那么,流分为这么几类(皆为抽象接口:

  • stream.Readable    可读流(需要实现_read方法,关注点在于对数据流读取的细节
  • stream.Writable    可写流(需要实现_write方法,关注点在于对数据流写入的细节
  • stream.Duplex       可读/写流(需要实现以上两接口,关注点为以上两接口的细节
  • stream.Transform 继承自Duplex(需要实现_transform方法,关注点在于对数据块的处理

简单来说:

  • .pipe() 的拥有者一定具备 Readable 流(并不局限于)能力,它拥有 'readable'/'data'/'end'/'close'/'error' 一系列事件可供订阅,也提供 .read()/.pause()/.resume()等一系列方法供调用;
  • .pipe() 的参数一定具备Writable 流(并不局限于 )能力,它拥有 'drain'/'pipe'/'unpipe'/'error'/'finish' 事件可供访问,也提供 .write()/.end() 等一系列方法供调用

什么鬼

有没有一丝丝焦虑?别急,做为一个说人话的低级码工,我会把Stream掰开了和您扯一扯的。

Stream类,在 Node.js的源码 里,是这么定义的:

var EE = require('events').EventEmitter;
var util = require('util');
util.inherits(Stream, EE); function Stream() {
EE.call(this);
}

可以看出,本质上,Stream是一个EventEmitter,那意味着它具备事件驱动的功能(.emit/.on...)。众所周知,“Node.js 就是基于V8的事件驱动平台”,实现了事件驱动的流式编程,具备了和Node一样的异步回调的特征。

比如在 Readable 流中,有一个 readable 事件,在一个暂停的只读流中,只要有数据块准备好可读时,它就会被发送给订阅者(Readable 流有哪些呢?express中的 req,ftp或者mutli-form上传组件的req.part,系统中的标准输入 process.stdin等)。有了readable 事件,我们可以做个处理shell 命令输出的分析器之类的工具:

process.stdin.on('readable', function(){
var buf = process.stdin.read();
if(buf){
var data = buf.toString();
// parsing data ...
}
});

这样调用:

head -10 some.txt | node parser.js

对于 Readable 流,我们还可以订阅它的 data 和 end 事件,以获取数据块并在流枯竭时获得通知,如 经典socket示例 中那样:

req.on('connect', function(res, socket, head) {
socket.on('data', function(chunk) {
console.log(chunk.toString());
});
socket.on('end', function() {
proxy.close();
});
});

Readable流状态的切换

需要注意的是,Readable 流有两种状态:flowing mode(激流) 和 pause  mode(暂停)。前者根本停不下来,谁被pipe上了就马上不停的给;后者会暂停,直到下游显式的调用 Stream.read() 请求才读取数据块。Readable 流初始化时是 pause mode的。

这两种状态可以互为切换的,其中,

有以下任一行为,pause 转 flowing:

  • 对 Readable 流添加一个data事件订阅
  • 对 Readable 调用 .resume() 显式开启flowing
  • 调用 Readable 流的 .pipe(writable) ,桥接到一个 Writable 流上

有以下任一行为,flowing 转回 pause:

  • Readable 流还没有 pipe 到任何流上,可调 .pause() 暂停
  • Readable 流已经 pipe 到了流上,需 remove 掉所有 data 事件订阅,并且调用 .unpipe()方法逐一解除与下游流的关系

妙用

结合流的异步特性,我可以写出这样的应用:直接将 用户A 的输出桥接到 用户B 的页面上输出:

router.post('/post', function(req, res) {
var destination = req.headers['destination']; //发给谁
cache[destionation] = req;
//是的,并不返回,所以最好是个ajax请求
});

  

用户B请求的时候:

router.get('/inbox', function(req, res){
var user = req.headers['user'];
cache.find(user, function(err, previousReq){ //找到之前存的req
var form = new multiparty.Form();
form.parse(previousReq); // 有文件给我
form.on('part', function (part) {
part.pipe(res); //流式大法好:) part.on('error', function (err) {
console.log(err);
messaging.setRequestDone(uniqueID);
return res.end(err);
});
});
});
});

  

参考

  1. how to write node programs with streams: stream-handbook

更多文章请移步我的blog新地址: http://www.moye.me/

[Node.js] Node.js中的流的更多相关文章

  1. ASP.NET Core SignalR中的流式传输

    什么是流式传输? 流式传输是这一种以稳定持续流的形式传输数据的技术. 流式传输的使用场景 有些场景中,服务器返回的数据量较大,等待时间较长,客户端不得不等待服务器返回所有数据后,再进行相应的操作.这时 ...

  2. node.js中stream流中可读流和可写流的使用

    node.js中的流 stream 是处理流式数据的抽象接口.node.js 提供了很多流对象,像http中的request和response,和 process.stdout 都是流的实例. 流可以 ...

  3. 理解 Node.js 中 Stream(流)

    Stream(流) 是 Node.js 中处理流式数据的抽象接口. stream 模块用于构建实现了流接口的对象. Node.js 提供了多种流对象. 例如,对 HTTP 服务器的request请求和 ...

  4. Node.js数据流Stream之Readable流和Writable流

    一.前传 Stream在很多语言都会有,当然Node.js也不例外.数据流是可读.可写.或即可读又可写的内存结构.Node.js中主要包括Readable.Writable.Duplex(双工)和Tr ...

  5. 极简 Node.js 入门 - 4.5 双工流

    极简 Node.js 入门系列教程:https://www.yuque.com/sunluyong/node 本文更佳阅读体验:https://www.yuque.com/sunluyong/node ...

  6. 在Sublime Text 3 中安装SublimeLinter,Node.js进行JS&CSS代码校验

    转载自:http://www.wiibil.com/website/sublimelinter-jshint-csslint.html 在Sublime Text中安装SublimeLinter,No ...

  7. node.js获取url中的各个参数

    实例代码test.js var http=require('http'); var url=require('url'); var querystring=require('querystring') ...

  8. Node.js数据流Stream之Duplex流和Transform流

    Duplex流一个很好的例子是TCP套接字连接.需要实现_read(size)和_Write(data,encoding,callback)方法. var stream = require('stre ...

  9. Elasticsearch.js 发布 —— 在Node.js和浏览器中调用Elasticsearch(1)

    继PHP.Ruby.Python和Perl之后,Elasticsearch最近发布了Elasticsearch.js,Elasticsearch的JavaScript客户端库.可以在Node.js和浏 ...

随机推荐

  1. C++ URLDecode和URLEncode实现——仅限gb2312,非utf8

    转载--http://blog.163.com/zhangjie_0303/blog/static/9908270620148251658993/   #include <iostream&gt ...

  2. &lbrack;POJ2773&rsqb;&colon;Happy 2006

    传送门 同样是欧拉函数的基本应用. $\phi (N)$表示$[1,N]$中,$gcd(i,N)==1$的数的个数,同理,其也能表示$[K \times N+1,(K+1) \times N]$中$g ...

  3. c&num; winform进入窗口后在文本框里的默认焦点

    c# winform 设置winform进入窗口后在文本框里的默认焦点   进入窗口后默认聚焦到某个文本框,两种方法: ①设置tabindex 把该文本框属性里的tabIndex设为0,焦点就默认在这 ...

  4. GBDT基本理论及利用GBDT组合特征的具体方法(收集的资料)

    最近两天在学习GBDT,看了一些资料,了解到GBDT由很多回归树构成,每一棵新回归树都是建立在上一棵回归树的损失函数梯度降低的方向. 以下为自己的理解,以及收集到的觉着特别好的学习资料. 1.GBDT ...

  5. GridView 使用方法总结 (一)

    GridView 使用方法总结 (一) 下载全部代码 http://www.sufeinet.com/thread-431-1-1.html   原文件作者是:csdn.net的清清月儿 她的主页  ...

  6. php-fpm 启动不了 libiconv&period;so&period;2找不到

    service php-fpm start  Starting php-fpm /alidata/server/php/sbin/php-fpm: error while loading shared ...

  7. PHP按值合并数组

    /** * PHP按值合并数组 * */ function my_array_merge(&$array1, &$array2) { $result = Array(); foreac ...

  8. 200行Java代码搞定计算器程序

    发现了大学时候写的计算器小程序,还有个图形界面,能够图形化展示表达式语法树,哈哈;) 只有200行Java代码,不但能够计算加减乘除,还能够匹配小括号~ 代码点评: 从朴素的界面配色到简单易懂错误提示 ...

  9. &lbrack;Leetcode 39&rsqb;组合数的和Combination Sum

    [题目] Given a set of candidate numbers (candidates) (without duplicates) and a target number (target) ...

  10. c&plus;&plus; 第二章知识梳理

    2.1.c++语言概括 2.1.1)c++的产生 一个更好的c,由c演变而来 2.1.2)c++的特点 一是尽量兼容c,二是支持面向对象的方法.更安全,且简洁高效. 2.1.3~2.1.5 多数和C相 ...