Node.js中的流(Stream)介绍

什么是流?

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

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

pipe

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

伪代码的表现形式为:

复制代码 代码如下:

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

流的分类

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

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

简单来说:

1).pipe() 的拥有者一定具备 Readable 流(并不局限于)能力,它拥有 'readable'/'data'/'end'/'close'/'error' 一系列事件可供订阅,也提供 .read()/.pause()/.resume()等一系列方法供调用;
2).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:

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

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

1.Readable 流还没有 pipe 到任何流上,可调 .pause() 暂停
2.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);
            });
        });
    });
});

参考

how to write node programs with streams: stream-handbook

时间: 2015-03-27

Nodejs学习笔记之Stream模块

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

node.js中的fs.createWriteStream方法使用说明

方法说明: 返回一个WriteStream(输出流)对象(可写流). 语法: 复制代码 代码如下: fs.createWriteStream(path, [options]) 由于该方法属于fs模块,使用前需要引入fs模块(var fs= require("fs") ) 接收参数: path    文件路径 option (object) 参数包含以下属性: 复制代码 代码如下: { flags: 'w',     encoding: null,     mode: 0666 } op

浅析Node.js 中 Stream API 的使用

本文由浅入深给大家介绍node.js stream api,具体详情请看下文吧. 基本介绍 在 Node.js 中,读取文件的方式有两种,一种是用 fs.readFile ,另外一种是利用 fs.createReadStream 来读取. fs.readFile 对于每个 Node.js 使用者来说最熟悉不过了,简单易懂,很好上手.但它的缺点是会先将数据全部读入内存,一旦遇到大文件的时候,这种方式读取的效率就非常低下了. 而 fs.createReadStream 则是通过 Stream 来读取

node.js中的fs.createReadStream方法使用说明

方法说明: 返回一个readStream(文件读取流,输入流)对象.(可读流) 语法: 复制代码 代码如下: fs.createReadStream(path, [options]) 由于该方法属于fs模块,使用前需要引入fs模块(var fs= require("fs") ) 接收参数: path: (string) 欲读取的文件路径 options : (object) 数组对象包含以下属性 复制代码 代码如下: { flags: 'r',   encoding: null,  

深入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

Stream在node.js中是一个抽象的接口,基于EventEmitter,也是一种Buffer的高级封装,用来处理流数据.流模块便是提供各种API让我们可以很简单的使用Stream. 流分为四种类型,如下所示: Readable,可读流 Writable,可写流 Duplex,读写流 Transform,扩展的Duplex,可修改写入的数据 1.Readable可读流 通过stream.Readable可创建一个可读流,它有两种模式:暂停和流动. 在流动模式下,将自动从下游系统读取数据并使用

浅析Node.js的Stream模块中的Readable对象

我一直都很不愿意扯 nodejs 的流,因为从第一次看到它我就觉得它的设计实在是太恶心了.但是没办法,Stream 规范尚未普及,而且确实有很多东西都依赖了 nodejs 的流来实现的,所以我也只能捏着鼻子硬着头皮来扯一扯这又臭又硬的 nodejs 流对象了. nodejs 自带了一个叫 stream 的模块,引入它便可以得到一组流对象构造器.现在我只说最简单的 stream.Readable. 其实用过 nodejs 的几乎都接触过 Readable 的实例,只是平时没太在意而已.一个非常典型

Node.js Streams文件读写操作详解

Node.js 天生异步和事件驱动,非常适合处理 I/O 相关的任务.如果你在处理应用中 I/O 相关的操作,你可以利用 Node.js 中的流(stream).因此,我们先具体看看流,理解一下它们是怎么简化 I/O 操作的吧. 流是什么 流是 unix 管道,让你可以很容易地从数据源读取数据,然后流向另一个目的地. 简单来说,流不是什么特别的东西,它只是一个实现了一些方法的 EventEmitter .根据它实现的方法,流可以变成可读流(Readable),可写流(Writable),或者双向

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

node.js中 stream使用教程

这些日子跑去学了一下OC,但是还没有学成.离转行的时间还有很长,顺便回顾一下node的知识. 每种语言来来去去的人很多,但我就离不开node.我并不是使用它开发,只是使用js相对多一些,因此还是研究node比较好,stream在node的地位是很高的.闲时也来看看这个内容,在node的路上,我还是新手. 今天下载了nodeschool的课程看,其中有一个例子.(我修改了一点点) var concat = require('concat-stream'); var http = require('

node.js中stream流中可读流和可写流的实现与使用方法实例分析

本文实例讲述了node.js中stream流中可读流和可写流的实现与使用方法.分享给大家供大家参考,具体如下: node.js中的流 stream 是处理流式数据的抽象接口.node.js 提供了很多流对象,像http中的request和response,和 process.stdout 都是流的实例. 流可以是 可读的,可写的,或是可读可写的.所有流都是 events 的实例. 一.流的类型 node.js中有四种基本流类型: 1.Writable 可写流 (例:fs.createWriteS

node.js中cluster的使用教程

本文主要给大家介绍了关于node.js中cluster使用的相关教程,分享出来供大家参考学习,下面来看看详细的介绍: 一.使用NODE中cluster利用多核CPU var cluster = require('cluster'); var http = require('http'); var numCPUs = require('os').cpus().length; if (cluster.isMaster) { // 创建工作进程 for (var i = 0; i < numCPUs;

Node.js中你不可不精的Stream(流)

一.什么是Stream(流) 流(stream)在 Node.js 中是处理流数据的抽象接口(abstract interface). stream 模块提供了基础的API.使用这些API可以很容易地来构建实现流接口的对象.例如, HTTP 请求 和 process.stdout 就都是流的实例. 流可以是可读的.可写的,或是可读写的.注意,所有的流都是 EventEmitter 的实例. 二.流的类型 Node.js 中有四种基本的流类型: Readable - 可读的流 (例如 fs.cre

详解node.js中的npm和webpack配置方法

概述 Node.js用c++语言编写而成的,是一个基于chrome V8引擎的javascript运行环境,让javaScript的运行脱离浏览器服务端,可以使用javaScript语言书写服务器端代码 1.使用node来实现一个http服务器 下面创建了一个端口为8787的服务器.他与php,java等不同,像php本地还要基于阿帕奇服务器,node.js能用代码快速搭建一个服务器. // 引入http模块 var http = require("http"); // 调用http的

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

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

Node.js插件安装图文教程

Node.js简介 Node.js是一个Javascript运行环境(runtime).实际上它是对Google V8引擎进行了封装.V8引 擎执行Javascript的速度非常快,性能非常好.Node.js对一些特殊用例进行了优化,提供了替代的API,使得V8在非浏览器环境下运行得更好. Node.js是一个基于Chrome JavaScript运行时建立的平台, 用于方便地搭建响应速度快.易于扩展的网络应用.Node.js 使用事件驱动, 非阻塞I/O 模型而得以轻量和高效,非常适合在分布式

Node.js+Express配置入门教程详解

Node.js是一个Javascript运行环境(runtime).实际上它是对Google V8引擎进行了封装.V8引 擎执行Javascript的速度非常快,性能非常好.Node.js对一些特殊用例进行了优化,提供了替代的API,使得V8在非浏览器环境下运行得更好.Node.js是一个基于Chrome JavaScript运行时建立的平台, 用于方便地搭建响应速度快.易于扩展的网络应用.Node.js 使用事件驱动, 非阻塞I/O 模型而得以轻量和高效,非常适合在分布式设备上运行的数据密集型

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

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