/** * 视频转码任务队列 * * 用于控制转码任务的并发数,避免低性能 NAS 设备过载 */ interface QueuedTask { materialId: string videoSourceUrl: string needsCoverImage: boolean resolve: () => void reject: (error: Error) => void } class VideoTranscodeQueue { private queue: QueuedTask[] = [] private processing: Set = new Set() private queued: Set = new Set() private maxConcurrent: number constructor(maxConcurrent?: number) { // 默认并发数为 1,适合低性能 NAS 设备 // 可以通过环境变量配置:VIDEO_TRANSCODE_MAX_CONCURRENT const configuredMaxConcurrent = parseInt(process.env.VIDEO_TRANSCODE_MAX_CONCURRENT || '1', 10) const resolvedMaxConcurrent = maxConcurrent ?? configuredMaxConcurrent this.maxConcurrent = Number.isFinite(resolvedMaxConcurrent) && resolvedMaxConcurrent > 0 ? resolvedMaxConcurrent : 1 } /** * 添加转码任务到队列 */ async add( materialId: string, videoSourceUrl: string, needsCoverImage: boolean, taskFn: (materialId: string, videoSourceUrl: string, needsCoverImage: boolean, retryCount?: number) => Promise ): Promise { // 如果已经在处理中,直接返回 if (this.processing.has(materialId)) { console.log(`[TranscodeQueue] Material ${materialId} is already processing, skipping`) return } if (this.queued.has(materialId)) { console.log(`[TranscodeQueue] Material ${materialId} is already queued, skipping`) return } return new Promise((resolve, reject) => { this.queue.push({ materialId, videoSourceUrl, needsCoverImage, resolve, reject, }) this.queued.add(materialId) // 立即尝试处理队列 this.process(taskFn) }) } /** * 处理队列中的任务 */ private async process( taskFn: (materialId: string, videoSourceUrl: string, needsCoverImage: boolean) => Promise ): Promise { // 如果已达到最大并发数,等待 if (this.processing.size >= this.maxConcurrent) { console.log(`[TranscodeQueue] 已达到最大并发数 (${this.maxConcurrent}),等待...`) return } // 如果队列为空,返回 if (this.queue.length === 0) { return } // 取出队列中的第一个任务 const task = this.queue.shift() if (!task) { return } this.queued.delete(task.materialId) console.log(`[TranscodeQueue] 开始处理任务: ${task.materialId} (队列剩余: ${this.queue.length})`) // 标记为处理中 this.processing.add(task.materialId) // 执行任务(确保异步执行) Promise.resolve() .then(() => taskFn(task.materialId, task.videoSourceUrl, task.needsCoverImage)) .then(() => { console.log(`[TranscodeQueue] ✓ 任务完成: ${task.materialId}`) task.resolve() }) .catch((error) => { console.error(`[TranscodeQueue] ✗ 任务失败: ${task.materialId}`, error) task.reject(error) }) .finally(() => { // 从处理中移除 this.processing.delete(task.materialId) console.log(`[TranscodeQueue] 任务结束: ${task.materialId} (处理中: ${this.processing.size})`) // 继续处理下一个任务 this.process(taskFn) }) } /** * 获取队列状态 */ getStatus() { return { queueLength: this.queue.length, processingCount: this.processing.size, processingIds: Array.from(this.processing), queuedIds: Array.from(this.queued), maxConcurrent: this.maxConcurrent, } } } // 单例模式 export const transcodeQueue = new VideoTranscodeQueue()