如何在node.js中作为可读流发出/管道数组值？

Question

从数组创建可读流并将值管道传递到可写流的最佳方法是什么？我已经看到了使用setInterval的substack 示例,我可以使用0来成功实现间隔值,但是我在迭代大量数据并且每次触发gc都会减慢速度.

// Working with the setInterval wrapper
var arr = [1, 5, 3, 6, 8, 9];

function createStream () {
    var t = new stream;
    t.readable = true;
    var times = 0;
    var iv = setInterval(function () {
        t.emit('data', arr[times]);
        if (++times === arr.length) {
            t.emit('end');
            clearInterval(iv);
        }
    }
}, 0);

// Create the writable stream s
// ....

createStream().pipe(s);

我想做的是在没有setInterval的情况下发出值.也许使用这样的异步模块:

async.forEachSeries(arr, function(item, cb) {
    t.emit('data', item);
    cb();
}, function(err) {
 if (err) {
     console.log(err);
 }
 t.emit('end');
});

在这种情况下,我迭代数组并发出数据,但从不管道任何值.我已经看过shinout的ArrayStream,但我认为它是在v0.10之前创建的,它比我想要的更多开销.

Answer 1

您可以通过创建可读流并将值推入其中来解决此问题.

流是一个痛苦,但它往往更容易对与他们直接合作,而不是使用库.

要流式传输的字符串或缓冲区数组

如果你正在使用一个字符串或缓冲区数组,这将工作:

'use strict'
const Stream = require('stream')
const readable = new Stream.Readable()

readable.pipe(process.stdout)

const items = ['a', 'b', 'c']
items.forEach(item => readable.push(item))

// no more data
readable.push(null)

笔记:

• readable.pipe(process.stdout) 做两件事:将流放入"流动"模式并设置process.stdout可写流以从中接收数据 readable
• 该Readable#push方法用于可读流的创建者,而不是流消费者.
• 你需要做的Readable#push(null)是发出没有更多数据的信号.

要传输的非字符串数组

要从既不是字符串也不是缓冲区的事物数组中创建流,您需要可读流和可写流都处于"对象模式".在下面的示例中,我进行了以下更改:

• 使用初始化可读流 {objectMode: true}

• 而不是管道process.stdout,管道到对象模式下的简单可写流.

    'use strict'
    const Stream = require('stream')
  
    const readable = new Stream.Readable({objectMode: true})
  
    const writable = new Stream.Writable({objectMode: true})
    writable._write = (object, encoding, done) => {
      console.log(object)
  
      // ready to process the next chunk
      done()
    }
  
    readable.pipe(writable)
  
    const items = [1, 2, 3]
    items.forEach(item => readable.push(item))
  
    // end the stream
    readable.push(null)
  

业绩说明

数据来自哪里？如果它是流数据源,最好使用变换流来操纵流,而不是转换为数组或从数组转换.

Answer 2

从 Node 12.3 开始，您可以stream.Readable.from(iterable, [options])改为使用。

const { Readable } = require('stream');
const readableStream = Readable.from(arr);

Answer 3

tl; dr;

const items = [1,2,3]
const stream = new Readable({
  objectMode: true,
  read() {
    const item = items.pop()
    if (!item) {
      this.push(null);
      return;
    }
    this.push(item)
  },
})