Stream接口

来自《JavaScript 标准参考教程（alpha）》，by 阮一峰

目录

数据读写可以看作是事件模式（Event）的特例，不断发送的数据块好比一个个的事件。读数据是read事件，写数据是write事件，而数据块是事件附带的信息。Node 为这类情况提供了一个特殊接口Stream。

概述

概念

”数据流“（stream）是处理系统缓存的一种方式。操作系统采用数据块（chunk）的方式读取数据，每收到一次数据，就存入缓存。Node应用程序有两种缓存的处理方式，第一种是等到所有数据接收完毕，一次性从缓存读取，这就是传统的读取文件的方式；第二种是采用“数据流”的方式，收到一块数据，就读取一块，即在数据还没有接收完成时，就开始处理它。

第一种方式先将数据全部读入内存，然后处理，优点是符合直觉，流程非常自然，缺点是如果遇到大文件，要花很长时间，才能进入数据处理的步骤。第二种方式每次只读入数据的一小块，像“流水”一样，每当系统读入了一小块数据，就会触发一个事件，发出“新数据块”的信号。应用程序只要监听这个事件，就能掌握数据读取的进展，做出相应处理，这样就提高了程序的性能。

var fs = require('fs');

fs

.createReadStream('./data/customers.csv')

.pipe(process.stdout);

上面代码中，fs.createReadStream方法就是以”数据流“的方式读取文件，这可以在文件还没有读取完的情况下，就输出到标准输出。这显然对大文件的读取非常有利。

Unix操作系统从很早以前，就有“数据流”这个概念，它是不同进程之间传递数据的一种方式。管道命令（pipe）就起到在不同命令之间，连接数据流的作用。“数据流”相当于把较大的数据拆成很小的部分，一个命令只要部署了数据流接口，就可以把一个流的输出接到另一个流的输入。Node引入了这个概念，通过数据流接口为异步读写数据提供的统一接口，无论是硬盘数据、网络数据，还是内存数据，都可以采用这个接口读写。

数据流接口最大特点就是通过事件通信，具有readable、writable、drain、data、end、close等事件，既可以读取数据，也可以写入数据。读写数据时，每读入（或写入）一段数据，就会触发一次data事件，全部读取（或写入）完毕，触发end事件。如果发生错误，则触发error事件。

一个对象只要部署了数据流接口，就可以从它读取数据，或者写入数据。Node内部很多涉及IO处理的对象，都部署了Stream接口，下面就是其中的一些。

  • 文件读写
  • HTTP 请求的读写
  • TCP 连接
  • 标准输入输出

可读数据流

Stream 接口分成三类。

  • 可读数据流接口，用于对外提供数据。
  • 可写数据流接口，用于写入数据。
  • 双向数据流接口，用于读取和写入数据，比如Node的tcp sockets、zlib、crypto都部署了这个接口。

“可读数据流”用来产生数据。它表示数据的来源，只要一个对象提供“可读数据流”，就表示你可以从其中读取数据。

var Readable = require('stream').Readable;

var rs = new Readable();

rs.push('beep ');

rs.push('boop\n');

rs.push(null);

rs.pipe(process.stdout);

上面代码产生了一个可写数据流，最后将其写入标注输出。可读数据流的push方法，用来将数据输入缓存。rs.push(null)中的null，用来告诉rs，数据输入完毕。

“可读数据流”有两种状态：流动态和暂停态。处于流动态时，数据会尽快地从数据源导向用户的程序；处于暂停态时，必须显式调用stream.read()等指令，“可读数据流”才会释放数据。刚刚新建的时候，“可读数据流”处于暂停态。

三种方法可以让暂停态转为流动态。

  • 添加data事件的监听函数
  • 调用resume方法
  • 调用pipe方法将数据送往一个可写数据流

如果转为流动态时，没有data事件的监听函数，也没有pipe方法的目的地，那么数据将遗失。

以下两种方法可以让流动态转为暂停态。

  • 不存在pipe方法的目的地时，调用pause方法
  • 存在pipe方法的目的地时，移除所有data事件的监听函数，并且调用unpipe方法，移除所有pipe方法的目的地

注意，只移除data事件的监听函数，并不会自动引发数据流进入“暂停态”。另外，存在pipe方法的目的地时，调用pause方法，并不能保证数据流总是处于暂停态，一旦那些目的地发出数据请求，数据流有可能会继续提供数据。

每当系统有新的数据，该接口可以监听到data事件，从而回调函数。

var fs = require('fs');

var readableStream = fs.createReadStream('file.txt');

var data = '';

readableStream.setEncoding('utf8');

readableStream.on('data', function(chunk) {

  data+=chunk;

});

readableStream.on('end', function() {

  console.log(data);

});

上面代码中，fs模块的createReadStream方法，是部署了Stream接口的文件读取方法。该方法对指定的文件，返回一个对象。该对象只要监听data事件，回调函数就能读到数据。

除了data事件，监听readable事件，也可以读到数据。

var fs = require('fs');

var readableStream = fs.createReadStream('file.txt');

var data = '';

var chunk;

readableStream.setEncoding('utf8');

readableStream.on('readable', function() {

  while ((chunk=readableStream.read()) !== null) {

    data += chunk;

  }

});

readableStream.on('end', function() {

  console.log(data)

});

readable事件表示系统缓冲之中有可读的数据，使用read方法去读出数据。如果没有数据可读，read方法会返回null。

“可读数据流”除了read方法，还有以下方法。

  • Readable.pause() ：暂停数据流。已经存在的数据，也不再触发data事件，数据将保留在缓存之中，此时的数据流称为静态数据流。如果对静态数据流再次调用pause方法，数据流将重新开始流动，但是缓存中现有的数据，不会再触发data事件。
  • Readable.resume()：恢复暂停的数据流。
  • readable.unpipe()：从管道中移除目的地数据流。如果该方法使用时带有参数，会阻止“可读数据流”进入某个特定的目的地数据流。如果使用时不带有参数，则会移除所有的目的地数据流。

readable 属性

一个数据流的readable属性返回一个布尔值。如果数据流是一个仍然打开的可读数据流，就返回true，否则返回false。

read()

read方法从系统缓存读取并返回数据。如果读不到数据，则返回null。

该方法可以接受一个整数作为参数，表示所要读取数据的数量，然后会返回该数量的数据。如果读不到足够数量的数据，返回null。如果不提供这个参数，默认返回系统缓存之中的所有数据。

只在“暂停态”时，该方法才有必要手动调用。“流动态”时，该方法是自动调用的，直到系统缓存之中的数据被读光。

var readable = getReadableStreamSomehow();

readable.on('readable', function() {

  var chunk;

  while (null !== (chunk = readable.read())) {

    console.log('got %d bytes of data', chunk.length);

  }

});

如果该方法返回一个数据块，那么它就触发了data事件。

_read()

可读数据流的_read方法，可以将数据放入可读数据流。

var Readable = require('stream').Readable;

var rs = Readable();

var c = 97;

rs._read = function () {

  rs.push(String.fromCharCode(c++));

  if (c > 'z'.charCodeAt(0)) rs.push(null);

};

rs.pipe(process.stdout);

运行结果如下。

$ node read1.js

abcdefghijklmnopqrstuvwxyz

setEncoding()

调用该方法，会使得数据流返回指定编码的字符串，而不是缓存之中的二进制对象。比如，调用setEncoding('utf8')，数据流会返回UTF-8字符串，调用setEncoding('hex')，数据流会返回16进制的字符串。

setEncoding的参数是字符串的编码方法，比如utf8、ascii、base64等。

该方法会正确处理多字节的字符，而缓存的方法buf.toString(encoding)不会。所以如果想要从数据流读取字符串，应该总是使用该方法。

var readable = getReadableStreamSomehow();

readable.setEncoding('utf8');

readable.on('data', function(chunk) {

  assert.equal(typeof chunk, 'string');

  console.log('got %d characters of string data', chunk.length);

});

resume()

resume方法会使得“可读数据流”继续释放data事件，即转为流动态。

// 新建一个readable数据流

var readable = getReadableStreamSomehow();

readable.resume();

readable.on('end', function(chunk) {

  console.log('数据流到达尾部，未读取任务数据');

});

上面代码中，调用resume方法使得数据流进入流动态，只定义end事件的监听函数，不定义data事件的监听函数，表示不从数据流读取任何数据，只监听数据流到达尾部。

pause()

pause方法使得流动态的数据流，停止释放data事件，转而进入暂停态。任何此时已经可以读到的数据，都将停留在系统缓存。

// 新建一个readable数据流

var readable = getReadableStreamSomehow();

readable.on('data', function(chunk) {

  console.log('读取%d字节的数据', chunk.length);

  readable.pause();

  console.log('接下来的1秒内不读取数据');

  setTimeout(function() {

    console.log('数据恢复读取');

    readable.resume();

  }, 1000);

});

isPaused()

该方法返回一个布尔值，表示“可读数据流”被客户端手动暂停（即调用了pause方法），目前还没有调用resume方法。

var readable = new stream.Readable

readable.isPaused() // === false

readable.pause()

readable.isPaused() // === true

readable.resume()

readable.isPaused() // === false

pipe()

pipe方法是自动传送数据的机制，就像管道一样。它从“可读数据流”读出所有数据，将其写出指定的目的地。整个过程是自动的。

src.pipe(dst)

pipe方法必须在可读数据流上调用，它的参数必须是可写数据流。

var fs = require('fs');

var readableStream = fs.createReadStream('file1.txt');

var writableStream = fs.createWriteStream('file2.txt');

readableStream.pipe(writableStream);

上面代码使用pipe方法，将file1的内容写入file2。整个过程由pipe方法管理，不用手动干预，所以可以将传送数据写得很简洁。

pipe方法返回目的地的数据流，因此可以使用链式写法，将多个数据流操作连在一起。

a.pipe(b).pipe(c).pipe(d)

// 等同于

a.pipe(b);

b.pipe(c);

c.pipe(d);

下面是一个例子。

var fs = require('fs');

var zlib = require('zlib');

fs.createReadStream('input.txt.gz')

  .pipe(zlib.createGunzip())

  .pipe(fs.createWriteStream('output.txt'));

上面代码采用链式写法，先读取文件，然后进行压缩，最后输出。

下面的写法模拟了Unix系统的cat命令，将标准输出写入标准输入。

process.stdin.pipe(process.stdout);

当来源地的数据流读取完成，默认会调用目的地的end方法，就不再能够写入。对pipe方法传入第二个参数{ end: false }，可以让目的地的数据流保持打开。

reader.pipe(writer, { end: false });

reader.on('end', function() {

  writer.end('Goodbye\n');

});

上面代码中，目的地数据流默认不会调用end方法，只能手动调用，因此“Goodbye”会被写入。

unpipe()

该方法移除pipe方法指定的数据流目的地。如果没有参数，则移除所有的pipe方法目的地。如果有参数，则移除该参数指定的目的地。如果没有匹配参数的目的地，则不会产生任何效果。

var readable = getReadableStreamSomehow();

var writable = fs.createWriteStream('file.txt');

readable.pipe(writable);

setTimeout(function() {

  console.log('停止写入file.txt');

  readable.unpipe(writable);

  console.log('手动关闭file.txt的写入数据流');

  writable.end();

}, 1000);

上面代码写入file.txt的时间，只有1秒钟，然后就停止写入。

事件

下面代码中，s是一个readable数据流，它可以监听以下事件。

s.on('data', f);    // 收到新的数据时，data事件就会发生，触发f()

s.on('end', f);     // 数据读取完以后，不会再收到数据了，end事件发生，触发f()

s.on('error', f);   // 发生错误时，error事件发生，触发f()

s.readable          // => true if it is a readable stream that is still open

s.pause();          // Pause "data" events.  For throttling uploads, e.g.

s.resume();         // Resume again

（1）readable

readable事件在数据流能够向外提供数据时触发。

```javascript

var readable = getReadableStreamSomehow();

readable.on('readable', function() {

  // there is some data to read now

});

下面是一个例子。

process.stdin.on('readable', function () {

  var buf = process.stdin.read();

  console.dir(buf);

});

上面代码将标准输入的数据读出。

read方法接受一个整数作为参数，表示以多少个字节为单位进行读取。

process.stdin.on('readable', function () {

    var buf = process.stdin.read(3);

    console.dir(buf);

});

上面代码将以3个字节为单位进行输出内容。

（2）data

对于那些没有显式暂停的数据流，添加data事件监听函数，会将数据流切换到流动态，尽快向外提供数据。

var readable = getReadableStreamSomehow();

readable.on('data', function(chunk) {

  console.log('got %d bytes of data', chunk.length);

});

（3）end

无法再读取到数据时，会触发end事件。也就是说，只有当前数据被完全读取完，才会触发end事件，比如不停地调用read方法。

var readable = getReadableStreamSomehow();

readable.on('data', function(chunk) {

  console.log('got %d bytes of data', chunk.length);

});

readable.on('end', function() {

  console.log('there will be no more data.');

});

（4）close

数据源关闭时，close事件被触发。并不是所有的数据流都支持这个事件。

（5）error

当读取数据发生错误时，error事件被触发。

继承可读数据流接口

可读数据流又分成两种，一种是 pull 模式，自己拉数据，就好像用吸管吸水，只有你吸了，水才会上来；另一种是 push 模式，数据自动推送给你，就好像水从水龙头自动涌出来。如果监听data事件，那么自动激活 push 模式；如果自己从数据流读取数据，那就是在使用 pull 模式。

任何对象都可以部署可读数据流的接口。

var Readable = require('stream').Readable;

var util = require('util');

function MyObject(options) {

  if (! (this instanceof MyObject)) return new MyObject(options);

  if (! options) options = {};

  options.objectMode = true;

  Readable.call(this, options);

}

util.inherits(MyObject, Readable);

MyObject.prototype._read = function read() {

  var self = this;

  someMethodGetData(function(err, data) {

    if (err) self.emit('error', err);

    else self.push(data);

  });

};

上面代码中，构造函数MyObject继承了读数据流的接口。options.objectMode设为true，是为了设置数据流处理的是对象，而不是字符串或者 buffer。此外，还要在MyObject.prototype上部署_read方法，每当数据流要读取数据，就会调用这个方法。在这个方法里面，我们取到数据，使用stream.push(data)将数据放进数据流。

然后，MyObject的实例就可以使用“读数据流”的接口了。

var myObj = new MyObject();

myObj.on('data', function(data) {

  console.log(data);

});

上面是 push 模式，下面是 pull 模式。

var myObj = new MyObject();

var data = myObj.read();

myObj也可以暂停/恢复读数据。

myObj.pause();

setTimeout(function () {

  myObj.resume();

}, 5000);

实例： fs 模块的读数据流

fs模块的createReadStream方法，就可以创建一个读取数据的数据流。

var fs = require('fs');

var stream = fs.createReadStream('readme.txt');

stream.setEncoding('utf8');

上面代码创建了一个文本文件readme.txt的数据流。由于这个数据流会当作文本处理，所以要用setEncoding方法设定编码。

然后，监听data事件，获取每一个数据块；监听end事件，当数据传送结束，再统一处理。

var data = '';

stream.on('data', function(chunk) {

  data += chunk;

})

stream.on('end', function() {

  console.log('Data length: %d', data.length);

});

监听readable事件，也可以取得与监听data事件同样的效果。

var data = '';

stream.on('readable', function() {

  var chunk;

  while(chunk = stream.read()) {

    data += chunk;

  }

});

数据流还有pause和resume方法，可以暂停和恢复数据传送。

// 暂停

stream.pause();

// 1秒后恢复

setTimeout(stream.resume(), 1000);

注意，数据流新建以后，默认状态是暂停，只有指定了data事件的回调函数，或者调用了resume方法，数据才会开发发送。

如果要同时使用readable与data事件，可以像下面这样写。

stream.pause();

var pulledData = '';

var pushedData = '';

stream.on('readable', function() {

  var chunk;

  while(chunk = stream.read()) {

    pulledData += chunk;

  }

});

stream.on('data', function(chunk) {

  pushedData += chunk;

});

上面代码中，显式调用pause方法，会使得readable事件释放一个data事件，否则data监听无效。

如果觉得data事件和end事件写起来太麻烦，Stream 接口还提供了pipe方法，自动处理这两个事件。数据流通过pipe方法，可以方便地导向其他具有Stream接口的对象。

var fs = require('fs');

var zlib = require('zlib');

fs.createReadStream('wow.txt')

  .pipe(zlib.createGzip())

  .pipe(process.stdout);

上面代码先打开文本文件wow.txt，然后压缩，再导向标准输出。

fs.createReadStream('wow.txt')

  .pipe(zlib.createGzip())

  .pipe(fs.createWriteStream('wow.gz'));

上面代码压缩文件wow.txt以后，又将其写回压缩文件。

下面代码新建一个Stream实例，然后指定写入事件和终止事件的回调函数，再将其接到标准输入之上。

var stream = require('stream');

var Stream = stream.Stream;

var ws = new Stream;

ws.writable = true;

ws.write = function(data) {

  console.log("input=" + data);

}

ws.end = function(data) {

  console.log("bye");

}

process.stdin.pipe(ws);

调用上面的脚本，会产生以下结果。

$ node pipe_out.js

hello

input=hello

^d

bye

上面代码调用脚本下，键入hello，会输出input=hello。然后按下ctrl-d，会输出bye。使用管道命令，可以看得更清楚。

$ echo hello | node pipe_out.js

input=hello

bye

可写数据流

“可读数据流”用来对外释放数据，“可写数据流”则是用来接收数据。它允许你将数据写入某个目的地。它是数据写入的一种抽象，不同的数据目的地部署了这个接口以后，就可以用统一的方法写入。

以下是部署了可写数据流的一些场合。

  • 客户端的http requests
  • 服务器的http responses
  • fs write streams
  • zlib streams
  • crypto streams
  • tcp sockets
  • child process stdin
  • process.stdout, process.stderr

只要调用stream.write(o)，就能将数据写入可读数据流。stream.write(payload, callback)可以指定回调函数callback，一旦缓存中的数据释放（payload），就会调用这个回调函数。

部署“可写数据流”，必须继承stream.Writable，以及实现stream._write方法。下面是一个例子，数据库的写入接口部署“可写数据流”接口。

var Writable = require('stream').Writable;

var util = require('util');

module.exports = DatabaseWriteStream;

function DatabaseWriteStream(options) {

  if (! (this instanceof DatabaseWriteStream))

    return new DatabaseWriteStream(options);

  if (! options) options = {};

  options.objectMode = true;

  Writable.call(this, options);

}

util.inherits(DatabaseWriteStream, Writable);

DatabaseWriteStream.prototype._write = function write(doc, encoding, callback) {

  insertIntoDatabase(JSON.stringify(doc), callback);

};

上面代码中，_write方法执行实际的写入操作，它必须接受三个参数。

  • chunk：要写入的数据块
  • encoding：如果写入的是字符串，必须字符串的编码
  • callback：写入完成后或发生错误时的回调函数

下面是用法的例子。

var DbWriteStream = require('./db_write_stream');

var db = DbWriteStream();

var Thermometer = require('./thermometer');

var thermomether = Thermometer();

thermomether.on('data', function(temp) {

  db.write({when: Date.now(), temperature: temp});

});

下面是fs模块的可写数据流的例子。

var fs = require('fs');

var readableStream = fs.createReadStream('file1.txt');

var writableStream = fs.createWriteStream('file2.txt');

readableStream.setEncoding('utf8');

readableStream.on('data', function(chunk) {

  writableStream.write(chunk);

});

上面代码中，fs模块的createWriteStream方法针对特定文件，创建了一个“可写数据流”，本质上就是对写入操作部署了Stream接口。然后，“可写数据流”的write方法，可以将数据写入文件。

writable属性

writable属性返回一个布尔值。如果数据流仍然打开，并且可写，就返回true，否则返回false。

s.writeable

write()

write方法用于向“可写数据流”写入数据。它接受两个参数，一个是写入的内容，可以是字符串，也可以是一个stream对象（比如可读数据流）或buffer对象（表示二进制数据），另一个是写入完成后的回调函数，它是可选的。

s.write(buffer);          // 写入二进制数据

s.write(string, encoding) // 写入字符串，编码默认为utf-8

write方法返回一个布尔值，表示本次数据是否处理完成。如果返回true，就表示可以写入新的数据了。如果等待写入的数据被缓存了，就返回false，表示此时不能立刻写入新的数据。不过，返回false的情况下，也可以继续传入新的数据等待写入。只是这时，新的数据不会真的写入，只会缓存在内存中。为了避免内存消耗，比较好的做法还是等待该方法返回true，然后再写入。

var fs = require('fs');

var ws = fs.createWriteStream('message.txt');

ws.write('beep ');

setTimeout(function () {

  ws.end('boop\n');

}, 1000);

上面代码调用end方法，数据就不再写入了。

cork()，uncork()

cork方法可以强制等待写入的数据进入缓存。当调用uncork方法或end方法时，缓存的数据就会吐出。

setDefaultEncoding()

setDefaultEncoding方法用于将写入的数据编码成新的格式。它返回一个布尔值，表示编码是否成功，如果返回false就表示编码失败。

end()

end方法用于终止“可写数据流”。该方法可以接受三个参数，全部都是可选参数。第一个参数是最后所要写入的数据，可以是字符串，也可以是stream对象或buffer对象；第二个参数是写入编码；第三个参数是一个回调函数，finish事件发生时，会触发这个回调函数。

s.end()                  // 关闭可写数据流

s.end(buffer)            // 最后一段写入二进制数据，然后关闭可写数据流

s.end(str, encoding)     // 最后一段写入字符串，然后关闭可写数据流

下面是一个例子。

var file = fs.createWriteStream('example.txt');

file.write('hello, ');

file.end('world!');

上面代码会在数据写入结束时，在尾部写入“world！”。

调用end方法之后，再写入数据会报错。

var file = fs.createWriteStream('example.txt');

file.end('world!');

file.write('hello, '); // 报错

事件

（1）drain事件

writable.write(chunk)返回false以后，当缓存数据全部写入完成，可以继续写入时，会触发drain事件，表示缓存空了。

s.on('drain', f);

下面是一个例子。

function writeOneMillionTimes(writer, data, encoding, callback) {

  var i = 1000000;

  write();

  function write() {

    var ok = true;

    do {

      i -= 1;

      if (i === 0) {

        writer.write(data, encoding, callback);

      } else {

        ok = writer.write(data, encoding);

      }

    } while (i > 0 && ok);

    if (i > 0) {

      writer.once('drain', write);

    }

  }

}

上面代码是一个写入100万次的例子，通过drain事件得到可以继续写入的通知。

（2）finish事件

调用end方法时，所有缓存的数据释放，触发finish事件。该事件的回调函数没有参数。

var writer = getWritableStreamSomehow();

for (var i = 0; i < 100; i ++) {

  writer.write('hello, #' + i + '!\n');

}

writer.end('this is the end\n');

writer.on('finish', function() {

  console.error('all writes are now complete.');

});

（3）pipe事件

“可写数据流”调用pipe方法，将数据流导向写入目的地时，触发该事件。

该事件的回调函数，接受发出该事件的“可读数据流”对象作为参数。

var writer = getWritableStreamSomehow();

var reader = getReadableStreamSomehow();

writer.on('pipe', function(src) {

  console.error('something is piping into the writer');

  assert.equal(src, reader);

});

reader.pipe(writer);

（4）unpipe事件

“可读数据流”调用unpipe方法，将可写数据流移出写入目的地时，触发该事件。

该事件的回调函数，接受发出该事件的“可读数据流”对象作为参数。

var writer = getWritableStreamSomehow();

var reader = getReadableStreamSomehow();

writer.on('unpipe', function(src) {

  console.error('something has stopped piping into the writer');

  assert.equal(src, reader);

});

reader.pipe(writer);

reader.unpipe(writer);

（5）error事件

如果写入数据或pipe数据时发生错误，就会触发该事件。

该事件的回调函数，接受一个Error对象作为参数。

pipe 方法

你可能会问为什么数据库要部署“可写数据流”接口，而不是直接使用原始的写入接口。答案就是为了可以使用pipe方法。

var DbWriteStream = require('./db_write_stream');

var db = DbWriteStream();

var Thermometer = require('./thermometer');

var thermomether = Thermometer();

thermomether.pipe(db);

// 10秒后断开连接

setTimeout(function () {

  thermometer.unpipe(db);

}, 10e3);

当可读数据流与可写数据流通过readable.pipe(writable)结合在一起时，数据会自动调整到消费者的速率。在内部，pipe使用“可写数据流”的.write()方法的返回值，来决定是否是否暂停读数据：如果writable.write返回true，表明数据已经写入完毕，缓存已经空了；如果返回false，就表示可写数据流正在缓存写入的数据，这意味着可以读取数据。等到”可写数据流“排空，就会释放drain事件，告诉数据源可以恢复释放数据了。

转换数据流

转换数据流用于将可读数据流释放的数据，转换成另一种格式，然后再发给可写数据流。

下面的例子是将一个JavaScript对象的数据流，转为JSON字符串的数据流。

// json_encode_stream.js

var Transform = require('stream').Transform;

var inherits = require('util').inherits;

module.exports = JSONEncode;

function JSONEncode(options) {

  if ( ! (this instanceof JSONEncode))

    return new JSONEncode(options);

  if (! options) options = {};

  options.objectMode = true;

  Transform.call(this, options);

}

inherits(JSONEncode, Transform);

JSONEncode.prototype._transform = function _transform(obj, encoding, callback) {

  try {

    obj = JSON.stringify(obj);

  } catch(err) {

    return callback(err);

  }

  this.push(obj);

  callback();

};

上面代码中，_transform方法接受原始的JavaScript对象，将它们转为JSON字符串。

然后，可读数据流与可写数据流之间，就可以用转换数据流连起来。

var DbWriteStream = require('./db_write_stream');

var db = DbWriteStream();

var JSONEncodeStream = require('./json_encode_stream');

var json = JSONEncodeStream();

var Thermometer = require('../thermometer');

var thermometer = Thermometer();

thermometer.pipe(json).pipe(db);

HTTP请求

HTTP对象使用Stream接口，实现网络数据的读写。

var http = require('http');

var server = http.createServer(function (req, res) {

  // req is an http.IncomingMessage, which is a Readable Stream

  // res is an http.ServerResponse, which is a Writable Stream

  var body = '';

  // we want to get the data as utf8 strings

  // If you don't set an encoding, then you'll get Buffer objects

  req.setEncoding('utf8');

  // Readable streams emit 'data' events once a listener is added

  req.on('data', function (chunk) {

    body += chunk;

  });

  // the end event tells you that you have entire body

  req.on('end', function () {

    try {

      var data = JSON.parse(body);

    } catch (er) {

      // uh oh!  bad json!

      res.statusCode = 400;

      return res

最新文章

  1. JavaScript禁用回退
  2. JDK环境变量设置
  3. 161226、js日期格式化
  4. oc中的枚举定义
  5. Android Studio HelloWorld
  6. UVa11526 H(n)
  7. Task schedule
  8. 使用Docker官方的Django包【转】
  9. Suse 创建NFS共享目录
  10. Linux之uniq命令
  11. (八) Usb摄像头描述符解析
  12. 【XSY2666】排列问题 DP 容斥原理 分治FFT
  13. Jenkins日常运维笔记-重启数据覆盖问题、迁移、基于java代码发版(maven构建)
  14. Go语言_range（范围）理解
  15. JSON之Asp.net MVC C#对象转JSON，DataTable转JSON，List&lt;T&gt;转JSON,JSON转List&lt;T&gt;,JSON转C#对象
  16. bing词典
  17. 自然语言处理词向量模型-word2vec
  18. 【记录】尝试用QEMU模拟ARM开发板去加载并运行Uboot，kernel，rootfs【转】
  19. hdfs调优
  20. 按钮在执行frame动画的时候怎么响应触发事件？

热门文章

  1. path-sum leetcode C++
  2. CentOS部署多台服务器JDK（shell脚本部署）
  3. 动手个性化设置自己的 IntelliJ IDEA
  4. 远程设备管理opendx平台搭建-server，agent以及front实际搭建
  5. MySQL 默认隔离级别是RR，为什么阿里这种大厂会改成RC？
  6. ONVIF客户端中预置位设置代码实现过程
  7. 集合之Map接口
  8. PTA 最小堆插入元素和删除堆顶（无哨兵元素） (20分)
  9. Linux 下权限的管理
  10. 【linux系统】命令学习（五）linux三剑客 grep \ awk \ sed