Node.js数据流Stream之Readable流和Writable流-阿里云开发者社区

开发者社区> 社会主义> 正文

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

简介:
+关注继续查看

一、前传

Stream在很多语言都会有,当然Node.js也不例外。数据流是可读、可写、或即可读又可写的内存结构。Node.js中主要包括Readable、Writable、Duplex(双工)和Transform(变换)流。但是在学这些之前先学会util模块中的一个从其他对象继承的功能.

util模块提供了util.inherits()方法来允许你创建一个继承另一个对象的prototype(原形)方法的对象。当创建一个新对象时,prototype方法自动被使用。

util.inherits(constructor,superconstructor)原形constructor被设定为原形superConstructor,并在一个新的对象被创建时执行。可以通过使用constructor.super_属性从自定义对象的构造函数访问supercontructor.

二、Readable流

有的前传util模块从其他对象继承的功能的了解,Readable就很好理解了.主要它包含以下方法和事件。

1.事件:

readable:在数据块可以从流中读取的时候发出。

data:类似readable,不同之处在于,当数据的事件处理程序被连接时,流被转变成流动的模式,并且数据处理程序被连续的调用,直到所有数据都被用尽

end:当数据不再被提供时由流发出

close:当底层资源,如文件,已关闭时发出。

error:在接收数据中出错是发出。

2.方法:

read([size]):从流中读数据.数据可以是String、Buffer、null(下面代码会有),当指定size,那么只读仅限于那个字节数

setEncoding(encoding):设置read()请求读取返回String时使用的编码

pause():暂停从该对象发出的data事件

resume():恢复从该对象发出的data事件

pipe(destination,[options]):把这个流的输出传输到一个由deatination(目的地)指定的Writable流对象。options是一个js对象.例如:{end:true}当Readable结束时就结束Writable目的地。

unpipe([destination]):从Writale目的地断开这一对象。

3.demo:


var stream = require('stream');
var util = require('util');
util.inherits(Answers, stream.Readable);
function Answers(opt) {
  stream.Readable.call(this, opt);
  this.quotes = ["yes", "no", "maybe"];
  this._index = 0;
}
Answers.prototype._read = function() {
  if (this._index > this.quotes.length){
    this.push(null);
  } else {
    this.push(this.quotes[this._index]);
    this._index += 1;
  }
};
var r = new Answers();
console.log("Direct read: " + r.read().toString());
r.on('data', function(data){
  console.log("Callback read: " + data.toString());
});
r.on('end', function(data){
  console.log("No more answers.");
});
r.on('readable',function(data)
{
   console.log('readable');
});

"C:\Program Files (x86)\JetBrains\WebStorm 11.0.3\bin\runnerw.exe" F:\nodejs\node.exe stream_read.js
Direct read: yes
readable
Callback read: no
Callback read: maybe
readable
readable
readable
No more answers.

Process finished with exit code 0


上面定义了一个通过util.inherits()继承Readable流的对象,从输出结果可以看到输出了3个字符串,但readable事件确执行了4次,其实前面也有写,read()可以是null,最后是push(null)了。

三、Writable流

有读就会有写,毕竟是可逆的,它和readable一样也有一些事件和方法

1.方法

write(chunk,[encoding],[callback]):将数据写入流。chunk(数据块)中包含要写入的数据,encoding指定字符串的编码,callback指定当数据已经完全刷新时执行的一个回调函数。如果成功写入,write()返回true.

end([chunk],[encoding],[callback]):与write()相同,它把Writable对象设为不再接受数据的状态,并发送finish事件。

2.事件

drain:在write()调用返回false后,当准备好开始写更多数据时,发出此事件通知监视器。

finish:当end()在Writable对象上调用,所以数据被刷新,并不会有更多的数据被接受时触发

pipe:当pipe()方法在Readable流上调用,已添加此writable为目的地时发出

unpipe:当unpipe()方法被调用,以删除Writable为目的地时发出。

3.demo


var stream = require('stream');
var util = require('util');
util.inherits(Writer, stream.Writable);
function Writer(opt) {
  stream.Writable.call(this, opt);
  this.data = new Array();
}
Writer.prototype._write = function(data, encoding, callback) {
  this.data.push(data.toString('utf8'));
  console.log("Adding: " + data);
  callback();
};
var w = new Writer();
for (var i=1; i<=5; i++){  
  w.write("Item" + i, 'utf8');
}
w.end("ItemLast");
console.log(w.data);

"C:\Program Files (x86)\JetBrains\WebStorm 11.0.3\bin\runnerw.exe" F:\nodejs\node.exe stream_write.js
Adding: Item1
Adding: Item2
Adding: Item3
Adding: Item4
Adding: Item5
Adding: ItemLast
[ 'Item1', 'Item2', 'Item3', 'Item4', 'Item5', 'ItemLast' ]

Process finished with exit code 0

四、把Readable流用管道输送到Writable流

上面也介绍了Readable流pipe()方法,这个主要是来测试


var stream = require('stream');
var util = require('util');
util.inherits(Reader, stream.Readable);
util.inherits(Writer, stream.Writable);
function Reader(opt) {
  stream.Readable.call(this, opt);
  this._index = 1;
}
Reader.prototype._read = function(size) {
  var i = this._index++;
  if (i > 10){
    this.push(null);
  } else {
    this.push("Item " + i.toString());
  }
};
function Writer(opt) {
  stream.Writable.call(this, opt);
  this._index = 1;
}
Writer.prototype._write = function(data, encoding, callback) {
  console.log(data.toString());
  callback();
};

var r = new Reader();
var w = new Writer();
w.on('pipe',function(){
  console.log('pipe');
});
r.pipe(w);

"C:\Program Files (x86)\JetBrains\WebStorm 11.0.3\bin\runnerw.exe" F:\nodejs\node.exe stream_piped.js
pipe
Item 1
Item 2
Item 3
Item 4
Item 5
Item 6
Item 7
Item 8
Item 9
Item 10

Process finished with exit code 0

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

相关文章
【译】Databricks使用Spark Streaming和Delta Lake对流式数据进行数据质量监控介绍
本文主要对Databricks如何使用Spark Streaming和Delta Lake对流式数据进行数据质量监控的方法和架构进行了介绍,本文探讨了一种数据管理架构,该架构可以在数据到达时,通过主动监控和分析来检测流式数据中损坏或不良的数据,并且不会造成瓶颈。
1194 0
InnoDB的数据页结构
页是InnoDB存储引擎管理数据库的最小磁盘单位。页类型为B-tree node的页,存放的即是表中行的实际数据了。 InnoDB数据页由以下七个部分组成,如图所示: File Header(文件头)。
757 0
Node.js爬取豆瓣数据
一直自以为自己vue还可以,一直自以为webpack还可以,今天在慕课逛node的时候,才发现,自己还差的很远。众所周知,vue-cli基于webpack,而webpack基于node,对node不了解,谈什么了解webpack。
1603 0
通过EMR Spark Streaming实时读取Tablestore数据
本文将介绍如何在E-MapReduce中实时流式的处理Tablestore中的数据。 场景设计 随着互联网的发展,企业中积累的数据越来越多,数据的背后隐藏着巨大的价值,在双十一这样的节日中,电子商务企业都会在大屏幕上实时显示订单总量,由于订单总量巨大,不可能每隔一秒就到数据库中进行一次SQL统计,此时就需要用到流计算,而传统的方法都是需要借助Kafka消息队列来做流式计算,数据订单需要写入数据库与Kafka中,Spark Streaming 消费来自Kafka中的订单信息。
3086 0
使用 Kafka + Spark Streaming + Cassandra 构建数据实时处理引擎
Apache Kafka 是一个可扩展,高性能,低延迟的平台,允许我们像消息系统一样读取和写入数据。我们可以很容易地在 Java 中使用 Kafka。 Spark Streaming 是 Apache Spark 的一部分,是一个可扩展、高吞吐、容错的实时流处理引擎。
2829 0
thymeleaf 传递数据到js变量
thymeleaf 传递数据到js变量 如何把控制器传来的model中的值传递给js变量呢? 需要以下两个: var message = [[${message}]] 1.
2861 0
【流数据与大屏DataV】如何使用DTS,Datahub,StreamCompute,RDS及DataV搭建流数据大屏
本文主要从数字化大屏的价值及实现两方面阐述了数字化大屏的制作过程。
3255 0
+关注
社会主义
从.Net到iOS,在撸的道路上越走越远,工作之余经营着博客园http://www.cnblogs.com/5ishare,欢迎小伙伴(妹子更好)一起交流,谈谈人生理想。作为经常从网上索取免费资料的一员,要有回报回报的思想,也让更多的人少走弯路.
237
文章
0
问答
文章排行榜
最热
最新
相关电子书
更多
《2021云上架构与运维峰会演讲合集》
立即下载
《零基础CSS入门教程》
立即下载
《零基础HTML入门教程》
立即下载