这里有新鲜出炉的 Node.JS 入门教程,程序狗速度看过来!
Node.js 是一个基于 Chrome JavaScript 运行时建立的一个平台, 用来方便地搭建快速的 易于扩展的网络应用 · Node.js 借助事件驱动, 非阻塞 I/O 模型变得轻量和高效, 非常适合 运行在分布式设备 的 数据密集型 的实时应用
这篇文章主要介绍了 Node.js 中的流(Stream)介绍, 本文讲解了什么是流、pipe 方法、流的分类、Readable 流状态的切换等内容, 需要的朋友可以参考下
什么是流?
说到流,就涉及到一个 * 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
来源: http://www.phperz.com/article/17/0425/272078.html