如何在 node.js 中发出/管道数组值作为可读流? [英] How to emit/pipe array values as a readable stream in node.js?

问题描述

从数组和管道值到可写流创建可读流的最佳方法是什么?我已经看到 substack 的 example 使用 setInterval 并且我可以成功地使用 0 作为间隔值，但是我正在迭代大量数据并每次触发 gc 都会减慢速度.

What is the best way to create a readable stream from an array and pipe values to a writable stream? I have seen substack's example using setInterval and I can implement that successfully using 0 for the interval value, but I am iterating over a lot of data and triggering gc every time is slowing things down.

// 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 的情况下发出值.也许像这样使用异步模块:

What I would like to do is emit values without the setInterval. Perhaps using the async module like this:

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 之前创建的，而且有点多开销比我正在寻找的要多.

In this case I iterate the array and emit data, but never pipe any values. I have already seen shinout's ArrayStream, but I think that was created before v0.10 and it is a bit more overhead than I am looking for.

推荐答案

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

You can solve this problem by creating a readable stream and pushing values into it.

流是一种痛苦，但通常更容易直接使用它们而不是使用库.

Streams are a pain, but it's often easier to work with them directly than to use libraries.

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

如果您正在处理字符串或缓冲区数组，这将起作用:

Array of strings or buffers to stream

If you're working with an array of strings or buffers, this will work:

'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) 以表示没有更多数据.

• readable.pipe(process.stdout) does two things: puts the stream into "flowing" mode and sets up the process.stdout writable stream to receive data from readable
• the Readable#push method is for the creator of the readable stream, not the stream consumer.
• You have to do Readable#push(null) to signal that there is no more data.

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

To make a stream from an array of things that are neither strings nor buffers, you need both the readable stream and the writable stream to be in "Object Mode". In the example below, I made the following changes:

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

• 不是通过管道传输到 process.stdout，而是通过管道传输到处于对象模式的简单可写流.

• Initialize the readable stream with {objectMode: true}

• Instead of piping to process.stdout, pipe to a simple writable stream that is in object mode.

  '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)

数据从哪里来?如果是流式数据源，最好使用转换流来操作流，而不是在数组之间进行转换.

Where is the data coming from? If it's a streaming data source, it's better to manipulate the stream using a transform stream than to convert to/from an array.

这篇关于如何在 node.js 中发出/管道数组值作为可读流?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！