node.js - 如何在异步调用后通过管道传输流而不丢失数据？

Question

在我的应用程序中，我希望能够执行以下步骤:

1. 获取读取流；
2. 等待异步函数完成；
3. 通过管道将流传输到destination1；
4. 等待另一个异步函数完成；
5. 通过管道将目标 1 传输到目标 2。

我期望以下内容:

1. 流处理仅在第 #5 步之后开始
2. 数据不会丢失
3. 当流处理结束时，整个逻辑完全解析(.on("finish"))。

在提出任何问题之前，先看一个代码示例:

return new Promise(resolve => {
    logger.debug("Creating a stream");
    const stream = fs.createReadStream("/home/username/dev/resources/ex.tar.bz2");

    setTimeout(() => {
        logger.debug("Attaching pipe 1");
        const pipe1 = stream.pipe(
            through(
                function(data) {
                    logger.info("DATA in PIPE 1");
                    this.queue(data);
                },
                function() {
                    logger.info("END in PIPE 1");
                    this.queue(null);
                }
            )
        );

        stream.pause(); // LINE 1

        setTimeout(() => {
            logger.debug("Attaching pipe 2");
            const pipe2 = pipe1.pipe(
                through(
                    function() {
                        logger.info("DATA in PIPE 2");
                    },
                    function() {
                        logger.info("END in PIPE 2");
                        resolve();
                    }
                )
            )

            pipe2.resume(); // LINE 2
        }, 1000);
    }, 1000);
});

在此代码中，如果 LINE 1 和 LINE 2 都被删除，则该代码不起作用(打印 PIPE 1 中的 DATA 和 END in PIPE 1，永远不会解析)因为:

• 附加目标 1 开始数据流；
• 如果我理解正确的话，当附加目的地 2 时，数据已被消耗。

如果 LINE 1 和 LINE 2 都存在，则代码似乎可以工作(打印PIPE 1 中的数据，PIPE 2 中的数据 ，在 PIPE 1 中结束，在 PIPE 2 中结束并解决)，因为:

• LINE 1 停止来自 stream 的数据流；
• 附加目标 2(有点令人困惑)不会从原始源启动流程；
• 第 2 行启动数据流。

根据 NodeJS 文档:

if there are piped destinations, then calling stream.pause() will not guarantee that the stream will remain paused once those destinations drain and ask for more data

这引出了我的主要问题:是否有可能可靠完全按照我尝试的方式实现这一点(在管道之间进行异步调用)？

奖励问题:

1. 我猜想使用管道的正确方法可能是确保在一次性构建整个管道之前完成所有必需的异步调用。 我的猜测正确吗？
2. 为什么附加目的地 2 不会触发流，而附加目的地 1 却会触发流？
3. 如果我用 pipe1.resume() 或 stream.resume() 替换第 2 行，代码同样可以正常工作。我想这可以扩展到无限数量的管道。 为什么我可以通过在任何管道上调用.resume()来恢复原始流程？这份简历与管道连接期间应该发生的简历有何不同(显然工作方式不同)？

Answer 1

您正在体验 Heisenberg's uncertainty principle 的 Node 流变体 - 观察流的行为会改变流的行为。

在执行任何其他操作之前，请删除 through Stream 的实现(虽然非常简单，但这本身会影响行为)。让我们使用内置的 Passthrough 流，我们知道它没有副作用:

logger.debug("Attaching pipe 1");
const pipe1 = new PassThrough();
stream.pipe(pipe1);
pipe1.on('data', data => logger.info('DATA in PIPE 1')); 
pipe1.on('end', () => logger.info('END in PIPE 1')); 


// ...

logger.debug("Attaching pipe 2");
const pipe2 = new PassThrough();
pipe1.pipe(pipe2);
pipe2.on('data', data => logger.info('DATA in PIPE 2')); 
pipe2.on('end', () => {
    logger.info('END in PIPE 2');
    resolve();
}); 

输出:

Creating a stream
Attaching pipe 1
DATA in PIPE 1
END in PIPE 1
Attaching pipe 2
END in PIPE 2

因此，在没有暂停/恢复语句的情况下，这是可行的(它不应该永远挂起，我不确定你为什么会看到这种行为)；但是，pipe2 中没有数据。它当然不会等待或缓冲任何东西。

问题是，通过附加一个 on('data') 处理程序(through 也这样做)，您可以通知流它有办法消耗数据 - 它不需要缓冲任何内容。当我们将管道添加到 pipe2 时，它确实立即开始管道传输 - 没有任何数据可供管道使用，因为我们已经消耗了它。

尝试注释掉pipe1的data处理程序:

//pipe1.on('data', data => logger.info('DATA in PIPE 1'));

现在我们得到了我们所期望的结果:

Creating a stream
Attaching pipe 1
Attaching pipe 2
DATA in PIPE 2
END in PIPE 1
END in PIPE 2

现在，当我们创建读取流时，它会立即开始读取(读入缓冲区)；我们附加pipe1，它立即开始管道数据(进入pipe1的内部缓冲区)；然后我们附加pipe2，它立即开始管道数据(进入pipe2的内部缓冲区)。您可以无限期地继续此操作，最终通过管道传输到写入流并将数据泵入磁盘或 HTTP 响应等。