在 Node.js 流中一直实现背压

Question

我刚刚创建了一个简单的可读可写流对，它可以与 pipe() 连接。我对创建背压和控制 Readable 中发生读取的速率感兴趣。但是，我对如何实际实现这一点或者是否可以使用 Node.js 流有点困惑。 举个例子：

   const {Writable, Readable} = require('stream');

    function getWritable() {
        return new Writable({
            write: function (chunk, encoding, cb) {
                console.log(' => chunk => ', String(chunk));
                setTimeout(cb, 1500);
            }
        });
    }


    function getReadable(data) {
        return new Readable({
            encoding: 'utf8',
            objectMode: false,
            read: function (n) {
                // n =>  Number(16384)
                console.log('read is called');
                const d = data.shift();
                this.push(d ? String(d) : null);
            }
        });
    }


    const readableStrm = getReadable([1, 2, 3, 4, 5]);

    const piped = readableStrm.pipe(getWritable());

    piped.on('finish', function () {
        console.log('finish');
    });

如果你运行上面的代码，我们会看到'read is called'会被记录 5 次，远在 Writable 中的 write 方法看到数据之前。

我想做的是仅在 Writable 中的 write 方法触发其回调时才在 Readable 中调用 read()；当然 read() 方法必须先触发一次，但随后会等待可写对象准备好。

有没有办法控制read() 方法何时以某种方式在可读文件中触发？

最后，我真的不明白read()方法的目的是什么。

作为一个简单的例子，无论我从 read() 返回什么，我都无法让它停止阅读。 read 方法的意义何在，我们为什么要实现它？

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

const r = new Readable({
    objectMode: true,
    read: function (n) {
        console.log('is read');
        return false/null/true;  // nothing I return here makes a difference
    }
});


r.on('data', function (d) {
    console.log(d);
});


setInterval(function(){
    r.push('valid');
},1000);

Answer 1

Node.js 流非常强大，在缓冲和通过它们的数据流中提供了很多控制。

现在回答你的问题：

A) 您会看到所有数据首先被读取，然后写入被触发，因为您的数据流非常小。如果您使用千字节数据进行测试，您会看到读取流和写入流按顺序触发。顺序取决于读取流的缓冲容量和写入流创建的背压。例如，TCP 套接字读取流将比磁盘文件写入流快得多，从而产生背压。

B) 一个强大的读写流构造器选项是highWaterMark。您可以在 documentation 的 Buffering 部分了解更多信息。 highWaterMark 专门定义了可读/可写流的缓冲容量。默认值为 16kb。在上面的示例中，您可以将具有 highWaterMark 的可读流设置为 2 个字节，如下所示，您会看到差异（在实际情况中不需要，但您可以用于学习）。

function getReadable(data) {
    let i = 0;
    return new stream.Readable({
        highWaterWark: 2,  // <--- highWaterMark set to 2 byte. Preferably set it to 10 and increase the length of your input array.
        encoding: 'utf8',
        objectMode: false,
        read: function (n) {
            // n =>  Number(16384)
            console.log('read is called');
            const d = data[i++];
            this.push(d ? String(d) : null);
        }
    });
}

highWaterMark 的较小值将很快产生背压，并且可能对某些用例（例如读取网络数据）不利。

C) 还可以控制读流和写流中的数据流向。如果您的应用程序需要它，那么您可以从可写流控制可读流。特定方法readable.pause()、readable.read([size])、readable.resume()、readable.push(chunk[, encoding]) 和readable.unpipe([destination]) 允许您控制可读流（甚至是可写流）中数据的缓冲和流动。事实上，您甚至可以使用方法readable.unshift(chunk) 将数据从可写流推回可读流。 有类似的方法可以控制可写流中的数据。

D) read 和 write 方法是流实现的一部分。这些方法用于将流数据发送到底层资源，不应直接从您的应用程序数据中调用。基本上，它定义了流的设置。 （不确定我是否能够清楚地解释这一点）。

我强烈建议您阅读 Node.js documentation on streams。它会为您提供大量信息（比您从其他网站上的示例代码中获得的信息还要多）。

希望以上信息对你有所帮助。