Node.js计算密集型Readable流优化：FFmpeg拉取过慢问题

核心问题分析 #

你的场景里，Node.js主线程被node-canvas的CPU密集型绘制任务占满，导致流的push操作和ffmpeg子进程的stdin数据拉取被阻塞，必须等缓冲区攒到highWaterMark才会批量处理，没法做到边生成边消费。大缓冲区和硬编码超时的方案都不适合生产环境，下面是几个可靠的优化方向：

1. 用Worker线程剥离CPU密集型绘制任务 #

node-canvas的帧绘制是纯CPU操作，完全可以放到worker_threads里执行，释放主线程专门处理流的调度和ffmpeg的通信，从根源解决主线程阻塞问题。

修改步骤：

• 新建frame-worker.js文件负责帧绘制：

const { parentPort, workerData } = require('worker_threads');
const canvas = require('canvas');
const { DrawingService, BufferType, RenderingLibraryError } = require('vm-rendering-library');

async function drawFrame() {
  const { animationAssets, fullAnimationData, videoRenderingInput, frameIndex } = workerData;
  const drawingService = new DrawingService(animationAssets, fullAnimationData, videoRenderingInput, canvas);
  
  try {
    await drawingService.drawForFrame(frameIndex);
    const buffer = await drawingService.toBuffer(BufferType.RAW);
    parentPort.postMessage({ success: true, buffer });
  } catch (err) {
    parentPort.postMessage({ success: false, error: new RenderingLibraryError(err).message });
    drawingService.destroyStage();
  }
}

drawFrame();

• 修改FrameCreationStream，用Worker替代主线程绘制：

import { Worker } from 'worker_threads';
import canvas from 'canvas';
import {Readable} from 'stream';
import {IMAGE_STREAM_BUFFER_SIZE, PerformanceUtil, RenderingLibraryError, VideoRendererInput} from 'vm-rendering-backend-commons';
import {AnimationAssets, BufferType, DrawingService, FullAnimationData} from 'vm-rendering-library';

export class FrameCreationStream extends Readable {
    drawingService: DrawingService;
    endFrameIndex: number;
    currentFrameIndex: number = 0;
    startFrameIndex: number;
    frameTimer: [number, number];
    readTimer: [number, number];
    fullAnimationData: FullAnimationData;
    // 控制并发Worker数量，避免内存暴涨
    maxConcurrentWorkers = 3;
    activeWorkers = 0;

    constructor(animationAssets: AnimationAssets, fullAnimationData: FullAnimationData, videoRenderingInput: VideoRendererInput, frameTimer: [number, number]) {
        super({highWaterMark: IMAGE_STREAM_BUFFER_SIZE, objectMode: true});

        this.frameTimer = frameTimer;
        this.readTimer = PerformanceUtil.startTimer();
        this.fullAnimationData = fullAnimationData;
        this.startFrameIndex = Math.floor(videoRenderingInput.startFrameId);
        this.currentFrameIndex = this.startFrameIndex;
        this.endFrameIndex = Math.floor(videoRenderingInput.endFrameId);
        this.drawingService = new DrawingService(animationAssets, fullAnimationData, videoRenderingInput, canvas);
    }

    _read(): void {
        // 维持可控并发，同时最多生成maxConcurrentWorkers帧
        while (this.currentFrameIndex <= this.endFrameIndex && this.activeWorkers < this.maxConcurrentWorkers) {
            this.activeWorkers++;
            const frameIndex = this.currentFrameIndex;
            this.currentFrameIndex++;

            const worker = new Worker('./frame-worker.js', {
                workerData: {
                    animationAssets: this.drawingService.animationAssets,
                    fullAnimationData: this.fullAnimationData,
                    videoRenderingInput: this.drawingService.videoRenderingInput,
                    frameIndex
                }
            });

            worker.on('message', (msg) => {
                this.activeWorkers--;
                if (msg.success) {
                    this.push(msg.buffer);
                    // 主动触发下一轮生成，维持并发节奏
                    this._read();
                } else {
                    this.emit('error', new Error(msg.error));
                }
                worker.terminate();
            });

            worker.on('error', (err) => {
                this.activeWorkers--;
                this.emit('error', err);
                worker.terminate();
            });
        }

        // 所有帧生成完成且无活跃Worker时，结束流
        if (this.currentFrameIndex > this.endFrameIndex && this.activeWorkers === 0) {
            this.push(null);
            PerformanceUtil.logTimer(this.frameTimer, 'FRAME_STREAM');
        }
    }

    _destroy(): void {
        this.drawingService.destroyStage();
    }
}

2. 调整流的背压策略，让ffmpeg立即开始消费 #

1. 移除自定义read()方法：你注释说明这个方法仅用于调试，生产环境删掉它，避免干扰原生流的背压逻辑。
2. 适配objectMode的背压逻辑：objectMode下highWaterMark是对象数量，你可以把它设为5-10（远低于原25），配合Worker的并发生成，既能控制内存，又能让ffmpeg持续拿到数据。

3. 优化ffmpeg参数，减少初始等待 #

ffmpeg处理rawvideo时默认会做探测操作，导致初始阶段不消费数据，添加以下参数强制它立即处理输入：

// 在FfmpegService的args数组开头新增：
'-probesize', '32',
'-analyzeduration', '0',

这两个参数会跳过不必要的格式探测，让ffmpeg直接按你指定的BGRA格式处理输入，大幅缩短启动等待时间。

4. 禁用不必要的同步日志 #

你当前代码里的console.log和console.timeLog都是同步IO操作，会阻塞主线程，生产环境务必移除或替换成异步日志库（如pino），减少主线程额外负担。

内容的提问来源于stack exchange，提问作者flohall