NodeJs 专题
专题目录
您的位置:Js应用 > NodeJs专题 > Node.js 流
Node.js 流
作者:--    发布时间:2019-11-20 20:19:05
稳定性: 2 - 不稳定

流用于处理node.js中的流数据的抽象接口,在node里被不同的对象实现。例如,对http服务器的请求是流,process.stdout 是流。

流是可读的,可写的,或者是可读写的,所有的流是eventemitter的实例。

node.js访问流模块的方法如下所示:

const stream = require('stream');

你可以通过require('stream')加载stream基类。其中包括了readable流、writable流、duplex流和transform流的基类。

本文将分为3个部分进行介绍。

第一个部分解释了在你的程序中使用流时候需要了解的内容。如果你不用实现流式api,可以只看这个部分。

如果你想实现你自己的流,第二个部分解释了这部分api。这些api让你的实现更加简单。

第三个部分深入的解释了流是如何工作的,包括一些内部机制和函数,这些内容不要改动,除非你明确知道你要做什么。

面向流消费者的api

流可以是可读(readable),可写(writable),或者是可读可写的(duplex,双工)。

所有的流都是事件分发器(eventemitters),但是也有自己的方法和属性,这取决于他它们是可读(readable),可写(writable),或者兼具两者(duplex,双工)的。

如果流是可读写的,则它实现了下面的所有方法和事件。因此,这个部分api完全阐述了duplextransform流,即便他们的实现有所不同。

没有必要为了消费流而在你的程序里实现流的接口。如果你正在你的程序里实现流接口,请同时参考下面的流实现程序api

基本所有的node程序,无论多简单,都会使用到流。以下是一个使用流的例子:

javascript
var http = require('http');

var server = http.createserver(function (req, res) {
  // req is an http.incomingmessage, which is 可读流(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 stream) emit 'data' 事件 once a 监听器(listener) is added
  req.on('data', function (chunk) {
    body += chunk;
  });

  // the end 事件 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.end('error: ' + er.message);
    }

    // write back something interesting to the user:
    res.write(typeof data);
    res.end();
  });
});

server.listen(1337);

// $ curl localhost:1337 -d '{}'
// object
// $ curl localhost:1337 -d '"foo"'
// string
// $ curl localhost:1337 -d 'not json'
// error: unexpected token o

类: stream.readable

可读流(readable stream)接口是对你正在读取的数据的来源的抽象。换句话说,数据来来自

可读流(readable stream)不会分发数据,直到你表明准备就绪。

可读流(readable stream) 有2种模式: 流动模式(flowing mode)暂停模式(paused mode)。流动模式(flowing mode)时,尽快的从底层系统读取数据并提供给你的程序。暂停模式(paused mode)时,你必须明确的调用stream.read()来读取数据。暂停模式(paused mode)是默认模式。

注意: 如果没有绑定数据处理函数,并且没有pipe()目标,流会切换到流动模式(flowing mode),并且数据会丢失。

可以通过下面几个方法,将流切换到流动模式(flowing mode)。

  • 添加一个['data' 事件][]事件处理器来监听数据.
  • 调用resume()方法来明确的开启数据流。
  • 调用pipe()方法来发送数据给writable.

可以通过以下方法来切换到暂停模式(paused mode):

  • 如果没有“导流(pipe)”目标,调用pause()方法.
  • 如果有“导流(pipe)”目标,移除所有的['data'事件][]处理函数,调用unpipe()方法移除所有的“导流(pipe)”目标。

注意:为了向后兼容考虑, 移除'data'事件监听器并不会自动暂停流。同样的,当有导流目标时,调用pause()并不能保证流在那些目标排空后,请求更多数据时保持暂停状态。

可读流(readable stream)例子包括:

事件: 'readable'

当一个数据块可以从流中读出,将会触发'readable'事件.`

某些情况下, 如果没有准备好,监听一个'readable'事件将会导致一些数据从底层系统读取到内部缓存。

javascript
var readble = getreadablestreamsomehow();
readable.on('readable', function() {
  // there is some data to read now
});

一旦内部缓存排空,一旦有更多数据将会再次触发readable事件。

事件: 'data'

  • chunk {buffer | string} 数据块

绑定一个data事件的监听器(listener)到一个未明确暂停的流,会将流切换到流动模式。数据会尽额能的传递。

如果你像尽快的从流中获取数据,以下的方法是最快的:

javascript
var readable = getreadablestreamsomehow();
readable.on('data', function(chunk) {
  console.log('got %d bytes of data', chunk.length);
});

事件: 'end'

如果没有更多的可读数据,将会触发这个事件。

注意:只有数据已经被完全消费,end事件才会触发。可以通过切换到流动模式(flowing mode)来实现,或者通过调用重复调用read()获取数据,直到结束。

javascript
    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.');
    });  

事件: 'close'

当底层资源(例如源头的文件描述符)关闭时触发。并不是所有流都会触发这个事件。

事件: 'error'

  • {error object}

当接收数据时发生错误触发。

readable.read([size])

  • size {number} 可选参数, 需要读入的数据量
  • 返回 {string | buffer | null}

read()方法从内部缓存中拉取数据。如果没有可用数据,将会返回null

如果传了size参数,将会返回相当字节的数据。如果size不可用,将会返回null

如果你没有指定size参数。将会返回内部缓存的所有数据。

这个方法仅能再暂停模式(paused mode)里调用。 流动模式(flowing mode)下这个方法会被自动调用直到内存缓存排空。

javascript
var readable = getreadablestreamsomehow();
readable.on('readable', function() {
  var chunk;
  while (null !== (chunk = readable.read())) {
    console.log('got %d bytes of data', chunk.length);
  }
});

如果这个方法返回一个数据块, 它同时也会触发['data'事件][].

readable.setencoding(encoding)

  • encoding {string} 要使用的编码.
  • 返回:this

调用此函数会使得流返回指定编码的字符串,而不是buffer对象。例如,如果你调用readable.setencoding('utf8'),输出数据将会是utf-8编码,并且返回字符串。如果你调用readable.setencoding('hex'),将会返回2进制编码的数据。

该方法能正确处理多字节字符。如果不想这么做,仅简单的直接拉取缓存并调buf.tostring(encoding),可能会导致字节错位。因此,如果你想以字符串读取数据,请使用下述的方法:

javascript
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);
});

readable.resume()

  • 返回:this

这个方法让可读流(readable stream)继续触发data事件.

这个方法会将流切换到流动模式(flowing mode)。 如果你不想从流中消费数据,而想得到end事件,可以调用readable.resume()来打开数据流,如下所示:

javascript
var readable = getreadablestreamsomehow();
readable.resume();
readable.on('end', function(chunk) {
  console.log('got to the end, but did not read anything');
});

readable.pause()

  • 返回:this

这个方法会使得流动模式(flowing mode)的流停止触发data事件,切换到流动模式(flowing mode)。并让后续可用数据留在内部缓冲区中。

javascript
var readable = getreadablestreamsomehow();
readable.on('data', function(chunk) {
  console.log('got %d bytes of data', chunk.length);
  readable.pause();
  console.log('there will be no more data for 1 second');
  settimeout(function() {
    console.log('now data will start flowing again');
    readable.resume();
  }, 1000);
});

readable.ispaused()

  • 返回:boolean

这个方法返回readable是否被客户端代码明确的暂停(调用readable.pause())。

var readable = new stream.readable

readable.ispaused() // === false
readable.pause()
readable.ispaused() // === true
readable.resume()
readable.ispaused() // === false

readable.pipe(destination[, options])

  • destination {writable stream} 写入数据的目标
  • options {object} 导流(pipe)选项
    • end {boolean} 读取到结束符时,结束写入者。默认 = true

这个方法从可读流(readable stream)拉取所有数据,并将数据写入到提供的目标中。自动管理流量,这样目标不会快速的可读流(readable stream)淹没。

可以导流到多个目标。

javascript
var readable = getreadablestreamsomehow();
var writable = fs.createwritestream('file.txt');
// all the data from readable goes into 'file.txt'
readable.pipe(writable);

这个函数返回目标流, 因此你可以建立导流链:

javascript
var r = fs.createreadstream('file.txt');
var z = zlib.creategzip();
var w = fs.createwritestream('file.txt.gz');
r.pipe(z).pipe(w);

例如:模拟unix的cat命令:

javascript
process.stdin.pipe(process.stdout);

默认情况下,当源数据流触发end的时候调用end(),所以destination不可再写。传{ end:false }作为options,可以保持目标流打开状态。

这会让writer保持打开状态,可以在最后写入"goodbye":

javascript
reader.pipe(writer, { end: false });
reader.on('end', function() {
  writer.end('goodbye\n');
});

注意:process.stderrprocess.stdout直到进程结束才会关闭,无论是否指定它们。

readable.unpipe([destination])

  • destination {writable stream} 可选,指定解除导流的流

这个方法会解除之前调用pipe() 设置的钩子(pipe())。

如果没有指定destination,则所有的导流(pipe)都会被移除。

如果指定了destination,但是没有建立如果没有指定destination,则什么事情都不会发生。

javascript
var readable = getreadablestreamsomehow();
var writable = fs.createwritestream('file.txt');
// all the data from readable goes into 'file.txt',
// but only for the first second
readable.pipe(writable);
settimeout(function() {
  console.log('stop writing to file.txt');
  readable.unpipe(writable);
  console.log('manually close the file stream');
  writable.end();
}, 1000);

readable.unshift(chunk)

  • chunk {buffer | string} 数据块插入到读队列中

这个方法很有用,当一个流正被一个解析器消费,解析器可能需要将某些刚拉取出的数据“逆消费”,返回到原来的源,以便流能将它传递给其它消费者。

如果你在程序中必须经常调用stream.unshift(chunk) ,那你可以考虑实现transform来替换(参见下文api for stream implementors)。

javascript
// pull off a header delimited by \n\n
// use unshift() if we get too much
// call the callback with (error, header, stream)
var stringdecoder = require('string_decoder').stringdecoder;
function parseheader(stream, callback) {
  stream.on('error', callback);
  stream.on('readable', onreadable);
  var decoder = new stringdecoder('utf8');
  var header = '';
  function onreadable() {
    var chunk;
    while (null !== (chunk = stream.read())) {
      var str = decoder.write(chunk);
      if (str.match(/\n\n/)) {
        // found the header boundary
        var split = str.split(/\n\n/);
        header += split.shift();
        var remaining = split.join('\n\n');
        var buf = new buffer(remaining, 'utf8');
        if (buf.length)
          stream.unshift(buf);
        stream.removelistener('error', callback);
        stream.removelistener('readable', onreadable);
        // now the body of the message can be read from the stream.
        callback(null, header, stream);
      } else {
        // still reading the header.
        header += str;
      }
    }
  }
}

readable.wrap(stream)

  • stream {stream} 一个旧式的可读流(readable stream)

v0.10版本之前的node流并未实现现在所有流的api(更多信息详见下文“兼容性”部分)。

如果你使用的是旧的node库,它触发'data'事件,并拥有仅做查询用的pause()方法,那么你能使用wrap()方法来创建一个readable流来使用旧版本的流,作为数据源。

你应该很少需要用到这个函数,但它会留下方便和旧版本的node程序和库交互。

例如:

javascript
var oldreader = require('./old-api-module.js').oldreader;
var oreader = new oldreader;
var readable = require('stream').readable;
var myreader = new readable().wrap(oreader);

myreader.on('readable', function() {
  myreader.read(); // etc.
});

类: stream.writable

可写流(writable stream )接口是你正把数据写到一个目标的抽象。

可写流(writable stream )的例子包括:

writable.write(chunk[, encoding][, callback])

  • chunk {string | buffer} 准备写的数据
  • encoding {string} 编码方式(如果chunk 是字符串)
  • callback {function} 数据块写入后的回调
  • 返回: {boolean} 如果数据已被全部处理返回true

这个方法向底层系统写入数据,并在数据处理完毕后调用所给的回调。

返回值表示你是否应该继续立即写入。如果数据要缓存在内部,将会返回false。否则返回true

返回值仅供参考。即使返回false,你也可能继续写。但是写会缓存在内存里,所以不要做的太过分。最好的办法是等待drain事件后,再写入数据。

事件: 'drain'

如果调用writable.write(chunk)返回 false,drain事件会告诉你什么时候将更多的数据写入到流中。

javascript
// write the data to the supplied 可写流(writable stream ) 1mm times.
// be attentive to back-pressure.
function writeonemilliontimes(writer, data, encoding, callback) {
  var i = 1000000;
  write();
  function write() {
    var ok = true;
    do {
      i -= 1;
      if (i === 0) {
        // last time!
        writer.write(data, encoding, callback);
      } else {
        // see if we should continue, or wait
        // don't pass the callback, because we're not done yet.
        ok = writer.write(data, encoding);
      }
    } while (i > 0 && ok);
    if (i > 0) {
      // had to stop early!
      // write some more once it drains
      writer.once('drain', write);
    }
  }
}

writable.cork()

强制缓存所有写入。

调用.uncork().end()后,会把缓存数据写入。

writable.uncork()

写入所有.cork()调用之后缓存的数据。

writable.setdefaultencoding(encoding)

  • encoding {string} 新的默认编码
  • 返回:boolean

给写数据流设置默认编码方式,如编码有效,则返回true ,否则返回false

writable.end([chunk][, encoding][, callback])

  • chunk {string | buffer} 可选,要写入的数据
  • encoding {string} 编码方式(如果chunk是字符串)
  • callback {function} 可选, stream结束时的回调函数

当没有更多的数据写入的时候调用这个方法。如果给出,回调会被用作finish事件的监听器。

调用end()后调用write()会产生错误。

javascript
// write 'hello, ' and then end with 'world!'
var file = fs.createwritestream('example.txt');
file.write('hello, ');
file.end('world!');
// writing more now is not allowed!

事件: 'finish'

调用end()方法后,并且所有的数据已经写入到底层系统,将会触发这个事件。

javascript
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.');
});

事件: 'pipe'

  • src {readable stream} 是导流(pipe)到可写流的源流。

无论何时在可写流(writable stream )上调用pipe()方法,都会触发'pipe'事件,添加这个流到目标。

javascript
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);

事件: 'unpipe'

  • src {readable stream}未写入此可写的源流。

无论何时在可写流(writable stream )上调用unpipe()方法,都会触发'unpipe'事件,将这个流从目标上移除。

javascript
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);

事件: 'error'

  • {error object}

写或导流(pipe)数据时,如果有错误会触发。

类: stream.duplex

双工流(duplex streams)是同时实现了readablewritable 接口。用法详见下文。

双工流(duplex streams) 的例子包括:

类: stream.transform

转换流(transform streams)是双工duplex流,它的输出是从输入计算得来。 它实现了readablewritable接口. 用法详见下文.

转换流(transform streams)的例子包括:

流实现程序api

无论实现什么形式的流,模式都是一样的:

  1. 在你的子类中扩展适合的父类。 (util.inherits方法很有帮助)
  2. 在你的构造函数中调用父类的构造函数,以确保内部的机制初始化正确。
  3. 实现一个或多个方法,如下所列。

所扩展的类和要实现的方法取决于你要编写的流类。

use-case

class

方法(s) to implement

reading only

[readable](#stream_class_stream_readable_1)

[_read][]

writing only

[writable](#stream_class_stream_writable_1)

[_write][]

reading and writing

[duplex](#stream_class_stream_duplex_1)

[_read][], [_write][]

operate on written data, then read the result

[transform](#stream_class_stream_transform_1)

_transform, _flush

在你的代码里,千万不要调用流实现程序api里的方法。否则可能会引起消费流的程序副作用。

类: stream.readable

stream.readable是一个可被扩充的、实现了底层_read(size)方法的抽象类。

参照之前的流实现程序api查看如何在你的程序里消费流。以下内容解释了在你的程序里如何实现可读流(readable stream)。

example: 计数流

这是可读流(readable stream)的基础例子,它将从1至1,000,000递增地触发数字,然后结束:

javascript
var readable = require('stream').readable;
var util = require('util');
util.inherits(counter, readable);

function counter(opt) {
  readable.call(this, opt);
  this._max = 1000000;
  this._index = 1;
}

counter.prototype._read = function() {
  var i = this._index++;
  if (i > this._max)
    this.push(null);
  else {
    var str = '' + i;
    var buf = new buffer(str, 'ascii');
    this.push(buf);
  }
};

example: 简单协议 v1 (初始版)

和之前描述的parseheader函数类似,但它被实现为自定义流。注意这个实现不会将输入数据转换为字符串。

实际上,更好的办法是将他实现为transform流,使用下面的实现方法会更好:

javascript
// a parser for a simple data protocol.
// "header" is a json object, followed by 2 \n characters, and
// then a message body.
//
// 注意: this can be done more simply as a transform stream!
// using readable directly for this is sub-optimal.  see the
// alternative example below under transform section.

var readable = require('stream').readable;
var util = require('util');

util.inherits(simpleprotocol, readable);

  function simpleprotocol(source, options) {
  if (!(this instanceof simpleprotocol))
    return new simpleprotocol(source, options);

  readable.call(this, options;

  this._inbody = false;
  this._sawfirstcr = false;

  // source is 可读流(readable stream), such as a socket or file
  this._source = source;

  var self = this;
  source.on('end', function() {
    self.push(null);
  });

  // give it a kick whenever the source is readable
  // read(0) will not consume any bytes
  source.on('readable', function() {
    self.read(0);
  });

  this._rawheader = [];
  this.header = null;
}

simpleprotocol.prototype._read = function(n) {
  if (!this._inbody) {
    var chunk = this._source.read();

    // if the source doesn't have data, we don't have data yet.
    if (chunk === null)
      return this.push('');

    // check if the chunk has a \n\n
    var split = -1;
    for (var i = 0; i < chunk.length; i++) {
      if (chunk[i] === 10) { // '\n'
        if (this._sawfirstcr) {
          split = i;
          break;
        } else {
          this._sawfirstcr = true;
        }
      } else {
        this._sawfirstcr = false;
      }
    }

    if (split === -1) {
      // still waiting for the \n\n
      // stash the chunk, and try again.
      this._rawheader.push(chunk);
      this.push('');
    } else {
      this._inbody = true;
      var h = chunk.slice(0, split);
      this._rawheader.push(h);
      var header = buffer.concat(this._rawheader).tostring();
      try {
        this.header = json.parse(header);
      } catch (er) {
        this.emit('error', new error('invalid simple protocol data'));
        return;
      }
      // now, because we got some extra data, unshift the rest
      // back into the 读取队列 so that our consumer will see it.
      var b = chunk.slice(split);
      this.unshift(b);

      // and let them know that we are done parsing the header.
      this.emit('header', this.header);
    }
  } else {
    // from there on, just provide the data to our consumer.
    // careful not to push(null), since that would indicate eof.
    var chunk = this._source.read();
    if (chunk) this.push(chunk);
  }
};

// usage:
// var parser = new simpleprotocol(source);
// now parser is 可读流(readable stream) that will emit 'header'
// with the parsed header data.

new stream.readable([options])

  • options{object}
    • highwatermark{number} 停止从底层资源读取数据前,存储在内部缓存的最大字节数;默认=16kb, objectmode流是16.
    • encoding{string} 若指定,则buffer会被解码成所给编码的字符串,默认为null。
    • objectmode{boolean} 该流是否为对象的流。意思是说stream.read(n)返回一个单独的值,而不是大小为n的buffer。

readable的扩展类中,确保调用了readable的构造函数,这样才能正确初始化。

readable._read(size)

  • size{number} 异步读取的字节数

注意:实现这个函数,但不要直接调用。

这个函数不要直接调用。在子类里实现,仅能被内部的readable类调用。

所有可读流(readable stream) 的实现必须停供一个_read方法,从底层资源里获取数据。

这个方法以下划线开头,是因为对于定义它的类是内部的,不会被用户程序直接调用。你可以在自己的扩展类中实现。

当数据可用时,通过调用readable.push(chunk)将之放到读取队列中。再次调用_read,需要继续推出更多数据。

size参数仅供参考。调用“read”可以知道知道应当抓取多少数据;其余与之无关的实现,比如tcp或tls,则可忽略这个参数,并在可用时返回数据。例如,没有必要“等到”size个字节可用时才调用stream.push(chunk)。

readable.push(chunk[, encoding])

  • chunk {buffer | null | string} 推入到读取队列的数据块
  • encoding {string} 字符串块的编码。必须是有效的buffer编码,比如utf8或ascii。
  • 返回{boolean}是否应该继续推入

注意: 这个函数必须被 readable 实现者调用, 而不是可读流(readable stream)的消费者.

_read()函数直到调用push(chunk)后才能被再次调用。

readable类将数据放到读取队列,当'readable'事件触发后,被read()方法取出。push()方法会插入数据到读取队列中。如果调用了null,会触发数据结束信号 (eof)。

这个api被设计成尽可能地灵活。比如说,你可以包装一个低级别的,具备某种暂停/恢复机制,和数据回调的数据源。这种情况下,你可以通过这种方式包装低级别来源对象:

javascript
// source is an object with readstop() and readstart() 方法s,
// and an `ondata` member that gets called when it has data, and
// an `onend` member that gets called when the data is over.

util.inherits(sourcewrapper, readable);

function sourcewrapper(options) {
  readable.call(this, options);

  this._source = getlowlevelsourceobject();
  var self = this;

  // every time there's data, we push it into the internal buffer.
  this._source.ondata = function(chunk) {
    // if push() 返回 false, then we need to stop reading from source
    if (!self.push(chunk))
      self._source.readstop();
  };

  // when the source ends, we push the eof-signaling `null` chunk
  this._source.onend = function() {
    self.push(null);
  };
}

// _read will be called when the stream wants to pull more data in
// the advisory size 参数 is ignored in this case.
sourcewrapper.prototype._read = function(size) {
  this._source.readstart();
};

类: stream.writable

stream.writable是个抽象类,它扩展了一个底层的实现_write(chunk, encoding, callback)方法.

参考上面的流实现程序api,来了解在你的程序里如何消费可写流。下面内容介绍了如何在你的程序里实现可写流。

new stream.writable([options])

  • options {object}
    • highwatermark {number} 当write()返回false时的缓存级别。默认=16kb,objectmode流是16。
    • decodestrings {boolean} 传给_write()前是否解码为字符串。默认=true
    • objectmode {boolean}write(anyobj)是否是有效操作;如果为true,可以写任意数据,而不仅仅是buffer/string。默认=false

请确保writable类的扩展类中,调用构造函数以便缓冲设定能被正确初始化。

writable._write(chunk, encoding, callback)

  • chunk {buffer | string} 要写入的数据块。总是buffer, 除非decodestrings选项为false
  • encoding {string} 如果数据块是字符串,这个参数就是编码方式。如果是缓存,则忽略。注意,除非decodestrings被设置为false,否则这个数据块一直是buffer。
  • callback{函数} 当你处理完数据后调用这个函数 (错误参数为可选参数)。

所以可写流(writable stream ) 实现必须提供一个_write()方法,来发送数据给底层资源。

注意: 这个函数不能直接调用,由子类实现,仅内部可写方法可以调用。

使用标准的callback(error)方法调用回调函数,来表明写入完成或遇到错误。

如果构造函数选项中设定了decodestrings标识,则chunk可能会是字符串而不是buffer,encoding表明了字符串的格式。这种设计是为了支持对某些字符串数据编码提供优化处理的实现。如果你没有明确的设置decodestringsfalse,这样你就可以安不管encoding参数,并假定chunk一直是一个缓存。

该方法以下划线开头,是因为对于定义它的类来说,这个方法是内部的,并且不应该被用户程序直接调用。你应当在你的扩充类中重写这个方法。

writable._writev(chunks, callback)

  • chunks {array} 准备写入的数据块,每个块格式如下:{ chunk: ..., encoding: ... }.
  • callback {函数} 当你处理完数据后调用这个函数 (错误参数为可选参数)。

注意: 这个函数不能直接调用。由子类实现,仅内部可写方法可以调用.

这个函数的实现是可选的。多数情况下,没有必要实现。如果实现,将会在所有数据块缓存到写队列后调用。

类: stream.duplex

双工流(duplex stream)同时兼具可读和可写特性,比如一个tcp socket连接。

注意stream.duplex可以像readable或writable一样被扩充,实现了底层_read(sise) 和_write(chunk, encoding, callback) 方法的抽象类。

由于javascript并没有多重继承能力,因此这个类继承自readable,寄生自writable.从而让用户在双工扩展类中同时实现低级别的_read(n)方法和低级别的_write(chunk, encoding, callback)方法。

new stream.duplex(options)

  • options {object} 传递writable and readable构造函数,有以下的内容:
    • allowhalfopen {boolean} 默认=true。 如果设置为false,当写端结束的时候,流会自动的结束读端,反之亦然。
    • readableobjectmode {boolean} 默认=false。将objectmode设为读端的流,如果为true,将没有效果。
    • writableobjectmode {boolean} 默认=false。将objectmode设为写端的流,如果为true,将没有效果。

扩展自duplex的类,确保调用了父亲的构造函数,保证缓存设置能正确初始化。

类: stream.transform

转换流(transform class) 是双工流(duplex stream),输入输出端有因果关系,比如,zlib流或crypto流。

输入输出没有要求大小相同,块数量相同,到达时间相同。例如,一个hash流只会在输入结束时产生一个数据块的输出;一个zlib流会产生比输入小得多或大得多的输出。

转换流(transform class) 必须实现_transform()方法,而不是_read()_write()方法,也可以实现_flush()方法(参见如下)。

new stream.transform([options])

  • options {object} 传递给writable和readable构造函数。

扩展自转换流(transform class) 的类,确保调用了父亲的构造函数,保证缓存设置能正确初始化。

transform._transform(chunk, encoding, callback)

  • chunk {buffer | string} 准备转换的数据块。是buffer,除非decodestrings选项设置为false
  • encoding {string} 如果数据块是字符串, 这个参数就是编码方式,否则就忽略这个参数
  • callback {函数} 当你处理完数据后调用这个函数 (错误参数为可选参数)。

注意:这个函数不能直接调用。由子类实现,仅内部可写方法可以调用.

所有的转换流(transform class) 实现必须提供 _transform方法来接收输入,并生产输出。

_transform可以做转换流(transform class)里的任何事,处理写入的字节,传给接口的写端,异步i/o,处理事情等等。

调用transform.push(outputchunk)0次或多次,从这个输入块里产生输出,依赖于你想要多少数据作为输出。

仅在当前数据块完全消费后调用这个回调。

注意,输入块可能有,也可能没有对应的输出块。如果你提供了第二个参数,将会传给push方法。如下述的例子:

javascript
transform.prototype._transform = function (data, encoding, callback) {
  this.push(data);
  callback();
}

transform.prototype._transform = function (data, encoding, callback) {
  callback(null, data);
}

该方法以下划线开头,是因为对于定义它的类来说,这个方法是内部的,并且不应该被用户程序直接调用。你应当在你的扩充类中重写这个方法。

transform._flush(callback)

  • callback {函数} 当你处理完数据后调用这个函数 (错误参数为可选参数)

注意:这个函数不能直接调用。由子类实现,仅内部可写方法可以调用.

某些情况下,转换操作可能需要分发一点流最后的数据。例如,zlib流会存储一些内部状态,以便优化压缩输出。

有些时候,你可以实现_flush方法,它可以在最后面调用,当所有的写入数据被消费后,分发end告诉读端。和_transform一样,当刷新操作完毕, transform.push(chunk)为0次或更多次数。

该方法以下划线开头,是因为对于定义它的类来说,这个方法是内部的,并且不应该被用户程序直接调用。你应当在你的扩充类中重写这个方法。

事件: 'finish' and 'end'

finishend事件 分别来自writable和readable类。.end()事件结束后调用finish事件,所有的数据已经被_transform处理完毕,调用_flush后,所有的数据输出完毕,触发end

example:simpleprotocolparser v2

上面的简单协议分析例子列子可以通过使用高级别的transform流来实现,和parseheadersimpleprotocol v1列子类似。

在这个示例中,输入会被导流到解析器中,而不是作为参数提供。这种做法更符合node流的惯例。

javascript
var util = require('util');
var transform = require('stream').transform;
util.inherits(simpleprotocol, transform);

function simpleprotocol(options) {
  if (!(this instanceof simpleprotocol))
    return new simpleprotocol(options);

  transform.call(this, options);
  this._inbody = false;
  this._sawfirstcr = false;
  this._rawheader = [];
  this.header = null;
}

simpleprotocol.prototype._transform = function(chunk, encoding, done) {
  if (!this._inbody) {
    // check if the chunk has a \n\n
    var split = -1;
    for (var i = 0; i < chunk.length; i++) {
      if (chunk[i] === 10) { // '\n'
        if (this._sawfirstcr) {
          split = i;
          break;
        } else {
          this._sawfirstcr = true;
        }
      } else {
        this._sawfirstcr = false;
      }
    }

    if (split === -1) {
      // still waiting for the \n\n
      // stash the chunk, and try again.
      this._rawheader.push(chunk);
    } else {
      this._inbody = true;
      var h = chunk.slice(0, split);
      this._rawheader.push(h);
      var header = buffer.concat(this._rawheader).tostring();
      try {
        this.header = json.parse(header);
      } catch (er) {
        this.emit('error', new error('invalid simple protocol data'));
        return;
      }
      // and let them know that we are done parsing the header.
      this.emit('header', this.header);

      // now, because we got some extra data, emit this first.
      this.push(chunk.slice(split));
    }
  } else {
    // from there on, just provide the data to our consumer as-is.
    this.push(chunk);
  }
  done();
};

// usage:
// var parser = new simpleprotocol();
// source.pipe(parser)
// now parser is 可读流(readable stream) that will emit 'header'
// with the parsed header data.

类: stream.passthrough

这是transform流的简单实现,将输入的字节简单的传递给输出。它的主要用途是测试和演示。偶尔要构建某种特殊流时也会用到。

流: 内部细节

缓冲

可写流(writable streams )和可读流(readable stream)都会缓存数据到内部对象上,叫做_writablestate.buffer_readablestate.buffer

缓存的数据量,取决于构造函数是传入的highwatermark参数。

调用stream.push(chunk)时,缓存数据到可读流(readable stream)。在数据消费者调用stream.read()前,数据会一直缓存在内部队列中。

调用stream.write(chunk)时,缓存数据到可写流(writable stream)。即使write()返回false

流(尤其是pipe()方法)得目的是限制数据的缓存量到一个可接受的水平,使得不同速度的源和目的不会淹没可用内存。

stream.read(0)

某些时候,你可能想不消费数据的情况下,触发底层可读流(readable stream)机制的刷新。这种情况下可以调用stream.read(0),它总会返回null。

如果内部读取缓冲低于highwatermark,并且流当前不在读取状态,那么调用read(0)会触发一个低级_read调用。

虽然基本上没有必要这么做。但你在node内部的某些地方看到它确实这么做了,尤其是在readable流类的内部。

stream.push('')

推一个0字节的字符串或缓存 (不在object mode时)会发送有趣的副作用。 因为它是一个对stream.push()的调用,它将会结束reading进程。然而,它没有添加任何数据到可读缓冲区中,所以没有东西可供用户消费。

少数情况下,你当时没有提供数据,但你的流的消费者(或你的代码的其它部分)会通过调用stream.read(0)得知何时再次检查。在这种情况下,你可以调用 stream.push('')

到目前为止,这个功能唯一一个使用情景是在tls.cryptostream类中,但它将在node v0.12中被废弃。如果你发现你不得不使用stream.push(''),请考虑另一种方式。

和老版本的兼容性

v0.10版本前,可读流(readable stream)接口比较简单,因此功能和用处也小。

  • 'data'事件会立即开始触发,而不会等待你调用read()方法。如果你需要进行某些i/o来决定如何处理数据,那么你只能将数据块储存到某种缓冲区中以防它们流失。
  • pause()方法仅供参考,而不保证生效。这意味着,即便流处于暂停状态时,你仍然需要准备接收'data'事件。

在node v0.10中, 加入了下文所述的readable类。为了考虑向后兼容,添加了'data'事件监听器或resume()方法被调用时,可读流(readable stream)会切换到 "流动模式(flowing mode)"。其作用是,即便你不使用新的read()方法和'readable'事件,你也不必担心丢失'data'数据块。

大多数程序会维持正常功能。然而,下列条件下也会引入边界情况:

  • 没有添加 ['data'事件][]处理器
  • 从来没有调用resume()方法
  • 流从来没有被倒流(pipe)到任何可写目标上、

例如:

javascript
// warning!  broken!
net.createserver(function(socket) {

  // we add an 'end' 方法, but never consume the data
  socket.on('end', function() {
    // it will never get here.
    socket.end('i got your message (but didnt read it)\n');
  });

}).listen(1337);

v0.10版本前的node,流入的消息数据会被简单的抛弃。之后的版本,socket会一直保持暂停。

这种情形下,调用resume()方法来开始工作:

javascript
// workaround
net.createserver(function(socket) {

  socket.on('end', function() {
    socket.end('i got your message (but didnt read it)\n');
  });

  // start the flow of data, discarding it.
  socket.resume();

}).listen(1337);

可读流(readable stream)切换到流动模式(flowing mode),v0.10 版本前,可以使用wrap()方法将风格流包含在一个可读类里。

object mode

通常情况下,流仅操作字符串和缓存。

处于object mode的流,除了缓存和字符串,还可以可以读出普通javascript值。

在对象模式里,可读流(readable stream) 调用stream.read(size)总会返回单个项目,无论是什么参数。

在对象模式里, 可写流(writable stream ) 总会忽略传给stream.write(data, encoding)encoding参数。

特殊值null在对象模式里,依旧保持它的特殊性。也就说,对于对象模式的可读流(readable stream),stream.read()返回null意味着没有更多数据,同时stream.push(null)会告知流数据结束(eof)。

node核心不存在对象模式的流,这种设计只被某些用户态流式库所使用。

应该在你的子类构造函数里,设置objectmode。在过程中设置不安全。

对于双工流(duplex streams),objectmode可以用readableobjectmodewritableobjectmode分别为读写端分别设置。这些选项,被转换流(transform streams)用来实现解析和序列化。

javascript
var util = require('util');
var stringdecoder = require('string_decoder').stringdecoder;
var transform = require('stream').transform;
util.inherits(jsonparsestream, transform);

// gets \n-delimited json  string data, and emits the parsed objects
function jsonparsestream() {
  if (!(this instanceof jsonparsestream))
    return new jsonparsestream();

  transform.call(this, { readableobjectmode : true });

  this._buffer = '';
  this._decoder = new stringdecoder('utf8');
}

jsonparsestream.prototype._transform = function(chunk, encoding, cb) {
  this._buffer += this._decoder.write(chunk);
  // split on newlines
  var lines = this._buffer.split(/\r?\n/);
  // keep the last partial line buffered
  this._buffer = lines.pop();
  for (var l = 0; l < lines.length; l++) {
    var line = lines[l];
    try {
      var obj = json.parse(line);
    } catch (er) {
      this.emit('error', er);
      return;
    }
    // push the parsed object out to the readable consumer
    this.push(obj);
  }
  cb();
};

jsonparsestream.prototype._flush = function(cb) {
  // just handle any leftover
  var rem = this._buffer.trim();
  if (rem) {
    try {
      var obj = json.parse(rem);
    } catch (er) {
      this.emit('error', er);
      return;
    }
    // push the parsed object out to the readable consumer
    this.push(obj);
  }
  cb();
};
网站声明:
本站部分内容来自网络,如您发现本站内容
侵害到您的利益,请联系本站管理员处理。
联系站长
373515719@qq.com
关于本站:
编程参考手册