从多个可读流中读取时 Node.js 停止

Question

在创建流 (A)、创建另一个流 (B) 并读取流 (B) 之后，读取过程从流 (A) 停止。 我该如何解决这个问题？

Node.js v14.18.1

import * as readline from 'readline';
import { Readable } from 'stream';

async function  main() {

    const  streamA = Readable.from('a');
    const  readerA = readline.createInterface({
        input: streamA,
        crlfDelay: Infinity
    });

    var  stopCase = false;
    if (stopCase) {

        const  streamB = Readable.from('b');
        const  readerB = readline.createInterface({
            input: streamB,
            crlfDelay: Infinity
        });

        console.log('readB');
        for await (const line of readerB) {
            console.log(line);
        }
    }
    console.log(`readerA.closed = ${'closed' in readerA}`);

    console.log('readA');
    for await (const line of readerA) {
        console.log(line);
    }
    console.log('success');
}
main();

输出（stopCase=true）：

readB
b
readerA.closed = true
readA

输出（stopCase=false）：

readerA.closed = false
readA
a
success

Answer 1

问题是，一旦你这样做：

const readerA = readline.createInterface({
    input: streamA,
    crlfDelay: Infinity
});

然后，streamA 现在准备好流动，readerA 准备好在您点击事件循环后立即生成事件。当您进入stopCase 块并点击for await (const line of readerB) 时，这将允许streamA 流动，这将允许readerA 触发事件。

但是，当 readerA 事件触发时，您并没有在监听它们，因此它完成了它在您没有监听时拥有的 streamA 内容。

如果您在完成 stopCase 块之前不创建 readerA，您会看到它如何更好地工作。因为当您点击 stopCase 块内的 await 时，streamA 和 readerA 还没有流动。

这就是我所说的成长的痛苦，它是由于尝试将承诺添加到事件驱动的流上而引起的。如果您让流处于流动状态并且您打算使用await 来读取这些事件，但是您随后await 做出了其他承​​诺，那么当您还没有收听时，您在第一个流上的所有事件都会触发。它不知道您正在等待使用await。您将其设置为流动，以便解释器一到达事件循环，它就会开始流动，即使您没有使用await 收听。

我之前在自己的代码中遇到过这种情况，解决方案是在您即将使用 await 读取它或直到您拥有更传统的事件处理程序之前，不要将流设置为流动配置为侦听任何流动的事件。基本上，您不能同时配置两个流以与for await (...) 一起使用。配置一个流，将它与您的for await (...) 一起使用，然后配置另一个。并且，还要注意在处理 for await (...) 循环时使用的任何其他承诺。使用这种结构时有很多方法可以搞砸。

在我看来，如果流实际上被置于不同的状态以与 Promise 一起使用，它会更可靠地工作，因此它只会通过 Promise 接口流动。那么，这种事情就不会发生了。但是，我敢肯定，该实施也存在许多挑战。

例如，如果你这样做：

import * as readline from 'readline';
import { Readable } from 'stream';

async function main() {
    var stopCase = true;
    console.log(`stopCase = ${stopCase}`);
    if (stopCase) {

        const streamB = Readable.from('b');
        const readerB = readline.createInterface({
            input: streamB,
            crlfDelay: Infinity
        });

        console.log('readB');
        for await (const line of readerB) {
            console.log(line);
        }
    }
    const streamA = Readable.from('a');
    const readerA = readline.createInterface({
        input: streamA,
        crlfDelay: Infinity
    });
    console.log(`streamA flowing = ${streamA.readableFlowing}`);
    console.log(`readerA.closed = ${!!readerA.closed}`);

    console.log('readA');
    for await (const line of readerA) {
        console.log(line);
    }
    console.log('success');
}
main();

然后，你得到所有的输出：

stopCase = true
readB
b
streamA flowing = true
readerA.closed = false
readA
a
success

你永远不会得到console.log('success') 的原因可能是因为你点击了for await (const line of readerA) { ...} 循环并且它在没有更多数据的承诺上停止了。同时，nodejs 注意到进程中没有任何东西可以创建任何未来的事件，因此它退出了进程。

您可以在更简单的应用中看到同样的概念：

async function main() {
    await new Promise(resolve => {
        // do nothing
    });
    console.log('success');
}
main();

它等待一个永远不会完成的承诺，并且应用程序中没有任何创建事件的事件，因此 nodejs 只是关闭并记录success。