能否用Node.js PassThrough Stream实现子进程输出的50MB缓冲？

问题背景 #

我在Node.js中运行名为generator的子进程，需要读取并处理其输出。该子进程有时会以约50MB/s的速率输出大量数据，但多数时候生成数据的速率较慢。我的读取代码偶尔也会变慢，读取速率下降。整体而言，Node.js程序的读取速率高于子进程的输出速率，但双方的速率波动会导致偶发背压，进而拖慢子进程。

我希望在Node.js中缓存最多约50MB的子进程输出，尝试了以下代码但未看到明显改善，也不知道如何进行准确的基准测试：

/**
 * 
 * @param nodeInputStream
 * @returns {Promise<null>} returns when end of stream is reached
 */
async function readAndProcessStream(nodeInputStream)  {
    // implementation redundant
    return;
}

async function createProcessAndRead() {
    const childArgs = ["arg1", "arg2"];
    const programName = "my_program";
    console.log("Spawn with args: ", programName, childArgs.join(" "));
    const childProc = child_process.spawn(
        programName,
        childArgs,
        {
            stdio:["ignore", "pipe", "ignore"],
            detached: true
        }
    );

    const exitCodePromise = new Promise((resolve, reject) => {
        childProc.once('close', resolve);
    });

    // Try to make a 50MB buffer
    const bufferStream = new PassThrough({emitClose: true, highWaterMark: 50*1024*1024});
    childProc.stdout.pipe(bufferStream);
    await readAndProcessStream(bufferStream);
    // make sure to wait till the process really exists
    await exitCodePromise;
}

核心疑问 #

请问上述代码能否在子进程与流处理函数之间实现50MB的缓冲空间？若不能，正确的实现方式是什么？

解答 #

原代码无法实现预期效果 #

你当前的代码达不到50MB缓冲的目的，原因有两点：

• PassThrough的highWaterMark不是全局缓冲容量，它只是单个读/写操作的缓冲区阈值，Node.js流的默认机制是分块处理，这个参数不会让PassThrough累计缓存50MB数据。
• 子进程的stdout本身是可读流，自带默认缓冲（约64KB），直接用pipe连接PassThrough后，背压还是会直接传递给子进程，并没有额外的缓冲空间来吸收速率波动。

正确实现方式 #

要实现最多50MB的缓冲，核心是在子进程输出和处理逻辑之间加入一个可控容量的缓冲层，以下是两种可行方案：

方案1：自定义缓冲可读流

手动封装一个可读流，内部维护一个容量上限为50MB的队列，当队列未满时持续读取子进程数据，处理逻辑变慢时队列暂存数据，避免背压传递到子进程：

const { Readable } = require('stream');
const child_process = require('child_process');

async function readAndProcessStream(nodeInputStream) {
    // 示例处理逻辑，可替换为实际业务代码
    for await (const chunk of nodeInputStream) {
        // 模拟偶尔变慢的场景
        if (Math.random() < 0.01) {
            await new Promise(resolve => setTimeout(resolve, 100));
        }
        // 处理chunk数据...
    }
}

class BufferedReadable extends Readable {
    constructor(sourceStream, maxBufferSize) {
        super({ highWaterMark: 1024 * 1024 }); // 设置单块数据大小
        this.source = sourceStream;
        this.maxBufferSize = maxBufferSize;
        this.currentBufferSize = 0;
        this.bufferQueue = [];
        this.isSourceEnded = false;

        // 监听子进程输出数据
        this.source.on('data', (chunk) => {
            this.bufferQueue.push(chunk);
            this.currentBufferSize += chunk.length;

            // 队列未满则继续读取，满了就暂停子进程输出
            if (this.currentBufferSize < this.maxBufferSize) {
                this.source.resume();
            } else {
                this.source.pause();
            }

            this.push(chunk); // 将数据推送给处理逻辑
        });

        this.source.on('end', () => {
            this.isSourceEnded = true;
            this.push(null); // 标记流结束
        });

        // 处理逻辑消费数据后，检查是否可以恢复子进程读取
        this.on('drain', () => {
            this.currentBufferSize -= this.readableHighWaterMark;
            if (this.currentBufferSize < this.maxBufferSize && !this.isSourceEnded) {
                this.source.resume();
            }
        });
    }

    _read() {
        // 无需额外实现，内部逻辑自动处理
    }
}

async function createProcessAndRead() {
    const childArgs = ["arg1", "arg2"];
    const programName = "my_program";
    console.log("Spawn with args: ", programName, childArgs.join(" "));
    const childProc = child_process.spawn(
        programName,
        childArgs,
        {
            stdio:["ignore", "pipe", "ignore"],
            detached: true
        }
    );

    const exitCodePromise = new Promise((resolve) => {
        childProc.once('close', resolve);
    });

    // 创建50MB容量的缓冲流
    const bufferedStream = new BufferedReadable(childProc.stdout, 50 * 1024 * 1024);
    await readAndProcessStream(bufferedStream);
    await exitCodePromise;
}

方案2：使用第三方库简化实现

如果不想手动写流逻辑，可以用bl（BufferList）库，它支持设置最大缓冲容量，自动处理背压：

const child_process = require('child_process');
const bl = require('bl');

async function readAndProcessStream(bufferList) {
    // 获取全部缓冲数据或逐块处理
    const fullData = await bufferList;
    // 处理数据逻辑...
}

async function createProcessAndRead() {
    const childArgs = ["arg1", "arg2"];
    const programName = "my_program";
    console.log("Spawn with args: ", programName, childArgs.join(" "));
    const childProc = child_process.spawn(
        programName,
        childArgs,
        {
            stdio:["ignore", "pipe", "ignore"],
            detached: true
        }
    );

    const exitCodePromise = new Promise((resolve) => {
        childProc.once('close', resolve);
    });

    // 设置50MB最大缓冲
    const bufferList = bl({ maxLength: 50 * 1024 * 1024 });
    childProc.stdout.pipe(bufferList);
    await readAndProcessStream(bufferList);
    await exitCodePromise;
}

基准测试建议 #

要验证缓冲效果，可以这样做：

• 让子进程以稳定的50MB/s速率输出测试数据，在处理逻辑中模拟周期性慢处理（比如每100块数据暂停100ms）
• 监控子进程的CPU使用率和输出速率，若缓冲生效，子进程不会因为处理逻辑变慢而频繁暂停
• 用process.memoryUsage()监控Node.js进程的内存占用，确认缓冲容量控制在50MB左右

内容的提问来源于stack exchange，提问作者Tomáš Zato