395 lines
14 KiB
TypeScript
395 lines
14 KiB
TypeScript
/**
|
||
* 视频转码任务队列
|
||
*
|
||
* 用于控制转码任务的并发数，充分利用多核 CPU
|
||
*
|
||
* 优化说明：
|
||
* - 默认并发数：2（充分利用多核 CPU）
|
||
* - 可以通过环境变量配置：VIDEO_TRANSCODE_MAX_CONCURRENT
|
||
* - 支持队列管理，避免系统过载
|
||
* - 支持部署中断后自动恢复转码任务
|
||
*
|
||
* 并发数建议：
|
||
* - H.265 编码：2 并发（CPU 占用 60-80%）
|
||
* - H.265 编码：3-4 并发（高性能服务器，CPU 占用 80-100%）
|
||
* - H.264 编码：3-4 并发（CPU 占用 60-80%）
|
||
*/
|
||
|
||
interface QueuedTask {
|
||
materialId: string
|
||
videoSourceUrl: string
|
||
needsCoverImage: boolean
|
||
resolve: () => void
|
||
reject: (error: Error) => void
|
||
}
|
||
|
||
class VideoTranscodeQueue {
|
||
private queue: QueuedTask[] = []
|
||
private processing: Set<string> = new Set()
|
||
private maxConcurrent: number
|
||
private progress: Map<string, number> = new Map() // 转码进度：materialId -> 0-100
|
||
private stages: Map<string, 'preview' | 'clip' | 'cover'> = new Map() // 转码阶段：materialId -> stage
|
||
|
||
constructor(maxConcurrent?: number) {
|
||
// 默认并发数计算逻辑：
|
||
// 1. 优先使用环境变量配置
|
||
// 2. 如果没有配置，使用默认值 2（充分利用多核 CPU）
|
||
// 3. 可以根据实际负载情况调整到 3-4（高性能服务器）
|
||
const envConcurrent = process.env.VIDEO_TRANSCODE_MAX_CONCURRENT
|
||
if (envConcurrent) {
|
||
this.maxConcurrent = parseInt(envConcurrent, 10) || 2
|
||
} else if (maxConcurrent !== undefined) {
|
||
this.maxConcurrent = maxConcurrent
|
||
} else {
|
||
// 默认值：2 并发（充分利用多核 CPU，提升 CPU 使用率）
|
||
// 可以根据实际负载情况调整到 3-4（高性能服务器）
|
||
// 注意：使用 H.265 (libx265) 时，2 并发可以达到 60-80% CPU 使用率
|
||
this.maxConcurrent = 2
|
||
}
|
||
|
||
console.log(`[TranscodeQueue] Initialized with maxConcurrent: ${this.maxConcurrent}`)
|
||
}
|
||
|
||
/**
|
||
* 添加转码任务到队列
|
||
*/
|
||
async add(
|
||
materialId: string,
|
||
videoSourceUrl: string,
|
||
needsCoverImage: boolean,
|
||
taskFn: (materialId: string, videoSourceUrl: string, needsCoverImage: boolean, retryCount?: number) => Promise<void>
|
||
): Promise<void> {
|
||
// 如果已经在处理中，直接返回
|
||
if (this.processing.has(materialId)) {
|
||
console.log(`[TranscodeQueue] Material ${materialId} is already processing, skipping`)
|
||
return
|
||
}
|
||
|
||
// 检查是否已在队列中等待
|
||
const alreadyInQueue = this.queue.some(t => t.materialId === materialId)
|
||
if (alreadyInQueue) {
|
||
console.log(`[TranscodeQueue] Material ${materialId} is already in queue, skipping`)
|
||
return
|
||
}
|
||
|
||
console.log(`[TranscodeQueue] Adding task to queue: materialId=${materialId}, videoSourceUrl=${videoSourceUrl}, needsCoverImage=${needsCoverImage}`)
|
||
|
||
return new Promise((resolve, reject) => {
|
||
this.queue.push({
|
||
materialId,
|
||
videoSourceUrl,
|
||
needsCoverImage,
|
||
resolve,
|
||
reject,
|
||
})
|
||
|
||
console.log(`[TranscodeQueue] Task added to queue. Queue length: ${this.queue.length}, Processing: ${this.processing.size}/${this.maxConcurrent}`)
|
||
|
||
// 立即尝试处理队列
|
||
this.process(taskFn).catch((error) => {
|
||
console.error(`[TranscodeQueue] Error in process() for material ${materialId}:`, error)
|
||
// 如果 process 本身出错，从队列中移除任务并拒绝 Promise
|
||
const taskIndex = this.queue.findIndex(t => t.materialId === materialId)
|
||
if (taskIndex >= 0) {
|
||
const task = this.queue[taskIndex]
|
||
this.queue.splice(taskIndex, 1)
|
||
task.reject(error)
|
||
}
|
||
})
|
||
})
|
||
}
|
||
|
||
/**
|
||
* 处理队列中的任务
|
||
*/
|
||
private async process(
|
||
taskFn: (materialId: string, videoSourceUrl: string, needsCoverImage: boolean, retryCount?: number) => Promise<void>
|
||
): Promise<void> {
|
||
try {
|
||
// 如果已达到最大并发数，等待
|
||
if (this.processing.size >= this.maxConcurrent) {
|
||
return
|
||
}
|
||
|
||
// 如果队列为空，返回
|
||
if (this.queue.length === 0) {
|
||
return
|
||
}
|
||
|
||
// 取出队列中的第一个任务
|
||
const task = this.queue.shift()
|
||
if (!task) {
|
||
return
|
||
}
|
||
|
||
// 验证任务参数
|
||
if (!task.materialId || !task.videoSourceUrl) {
|
||
console.error(`[TranscodeQueue] Invalid task parameters:`, task)
|
||
task.reject(new Error('Invalid task parameters'))
|
||
// 继续处理下一个任务
|
||
this.process(taskFn).catch(err => console.error('[TranscodeQueue] Error processing next task:', err))
|
||
return
|
||
}
|
||
|
||
// 标记为处理中
|
||
this.processing.add(task.materialId)
|
||
|
||
console.log(`[TranscodeQueue] 🚀 Starting task for material ${task.materialId} (queue: ${this.queue.length} waiting, ${this.processing.size}/${this.maxConcurrent} processing)`)
|
||
|
||
// 执行任务（传递 retryCount=0 作为初始重试次数）
|
||
// 注意：这里不 await，让任务异步执行，队列继续处理下一个任务
|
||
taskFn(task.materialId, task.videoSourceUrl, task.needsCoverImage, 0)
|
||
.then(() => {
|
||
// 转码成功，清除进度信息
|
||
this.clearProgress(task.materialId)
|
||
console.log(`[TranscodeQueue] ✅ Task completed for material ${task.materialId}`)
|
||
task.resolve()
|
||
})
|
||
.catch((error) => {
|
||
// 转码失败，清除进度信息
|
||
this.clearProgress(task.materialId)
|
||
const errorMsg = error?.message || String(error)
|
||
console.error(`[TranscodeQueue] ❌ Task failed for material ${task.materialId}:`, errorMsg)
|
||
task.reject(error)
|
||
})
|
||
.finally(() => {
|
||
// 从处理中移除
|
||
this.processing.delete(task.materialId)
|
||
console.log(`[TranscodeQueue] Task removed from processing. Remaining: ${this.processing.size}/${this.maxConcurrent}`)
|
||
// 继续处理下一个任务
|
||
this.process(taskFn).catch(err => {
|
||
console.error('[TranscodeQueue] Error processing next task after completion:', err)
|
||
})
|
||
})
|
||
} catch (error) {
|
||
console.error('[TranscodeQueue] Error in process() method:', error)
|
||
// 即使 process 方法本身出错，也要继续处理队列
|
||
// 避免整个队列卡死
|
||
setTimeout(() => {
|
||
this.process(taskFn).catch(err => {
|
||
console.error('[TranscodeQueue] Error in retry process():', err)
|
||
})
|
||
}, 1000)
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 获取队列状态
|
||
*/
|
||
getStatus() {
|
||
return {
|
||
queueLength: this.queue.length,
|
||
processingCount: this.processing.size,
|
||
processingIds: Array.from(this.processing),
|
||
maxConcurrent: this.maxConcurrent,
|
||
progress: this.progress,
|
||
stages: this.stages,
|
||
waiting: this.queue.map(t => t.materialId),
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 更新转码进度
|
||
*/
|
||
updateProgress(materialId: string, progress: number, stage?: 'preview' | 'clip' | 'cover') {
|
||
this.progress.set(materialId, progress)
|
||
if (stage) {
|
||
this.stages.set(materialId, stage)
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 清除转码进度（转码完成或失败时）
|
||
*/
|
||
clearProgress(materialId: string) {
|
||
this.progress.delete(materialId)
|
||
this.stages.delete(materialId)
|
||
}
|
||
|
||
/**
|
||
* 获取队列详情（包含等待中的素材ID列表，方便前端展示"排队位置"）
|
||
*/
|
||
getDetailedStatus() {
|
||
return {
|
||
queue: this.queue.map((t) => t.materialId),
|
||
processing: Array.from(this.processing),
|
||
maxConcurrent: this.maxConcurrent,
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 恢复中断的转码任务
|
||
*
|
||
* 部署中断后的恢复流程：
|
||
* 1. 查询所有 processing 状态的视频，清理残留文件并重置为 pending
|
||
* 2. 查询所有 pending 状态的视频，按时间顺序重新入队
|
||
*
|
||
* @param taskFn 转码任务执行函数
|
||
* @param cleanupFn 清理残留文件的函数
|
||
*/
|
||
async recoverInterruptedTasks(
|
||
taskFn: (materialId: string, videoSourceUrl: string, needsCoverImage: boolean, retryCount?: number) => Promise<void>,
|
||
cleanupFn: (material: { id: string; videoPreview540Url: string | null; videoPreviewClipUrl: string | null; videoCoverUrl: string | null }) => Promise<void>
|
||
): Promise<{ recovered: number; requeued: number }> {
|
||
const { prisma } = await import('@/lib/prisma')
|
||
|
||
let recovered = 0
|
||
let requeued = 0
|
||
|
||
try {
|
||
// 1. 查询所有 processing 状态的视频（部署中断时留下的）
|
||
const processingMaterials = await prisma.material.findMany({
|
||
where: {
|
||
videoProcessingStatus: 'processing',
|
||
videoSourceUrl: { not: null },
|
||
deletedAt: null,
|
||
} as any,
|
||
select: {
|
||
id: true,
|
||
videoPreview540Url: true,
|
||
videoPreviewClipUrl: true,
|
||
videoCoverUrl: true,
|
||
videoSourceUrl: true,
|
||
} as any,
|
||
})
|
||
|
||
console.log(`[TranscodeQueue] Found ${processingMaterials.length} interrupted tasks (processing status)`)
|
||
|
||
// 2. 清理残留文件并重置状态为 pending
|
||
for (const material of processingMaterials) {
|
||
try {
|
||
// 提取字段以避免类型错误
|
||
const materialId = String(material.id)
|
||
const preview540Url = material.videoPreview540Url ? String(material.videoPreview540Url) : null
|
||
const previewClipUrl = material.videoPreviewClipUrl ? String(material.videoPreviewClipUrl) : null
|
||
const coverUrl = material.videoCoverUrl ? String(material.videoCoverUrl) : null
|
||
|
||
// 清理残留文件（类型转换以匹配 cleanupFn 的期望）
|
||
await cleanupFn({
|
||
id: materialId,
|
||
videoPreview540Url: preview540Url,
|
||
videoPreviewClipUrl: previewClipUrl,
|
||
videoCoverUrl: coverUrl,
|
||
} as any)
|
||
|
||
// 重置状态为 pending，清除错误信息
|
||
await prisma.material.update({
|
||
where: { id: materialId },
|
||
data: {
|
||
videoProcessingStatus: 'pending',
|
||
transcodeLastError: null,
|
||
} as any,
|
||
})
|
||
|
||
recovered++
|
||
console.log(`[TranscodeQueue] Recovered task: ${material.id}`)
|
||
} catch (error) {
|
||
console.error(`[TranscodeQueue] Failed to recover task ${material.id}:`, error)
|
||
}
|
||
}
|
||
|
||
// 3. 查询所有 pending 状态的视频，按时间排序
|
||
const pendingMaterials = await prisma.material.findMany({
|
||
where: {
|
||
videoProcessingStatus: 'pending',
|
||
videoSourceUrl: { not: null },
|
||
deletedAt: null,
|
||
} as any,
|
||
orderBy: { createdAt: 'asc' }, // 按创建时间排序，保持原始队列顺序
|
||
select: {
|
||
id: true,
|
||
videoSourceUrl: true,
|
||
videoCoverUrl: true,
|
||
} as any,
|
||
})
|
||
|
||
console.log(`[TranscodeQueue] Found ${pendingMaterials.length} pending tasks to requeue`)
|
||
|
||
// 4. 按顺序重新入队（但不立即执行，等待队列调度）
|
||
for (const material of pendingMaterials) {
|
||
try {
|
||
// 提取字段以避免类型错误
|
||
const materialId = String(material.id)
|
||
const videoSourceUrl = material.videoSourceUrl ? String(material.videoSourceUrl) : null
|
||
const needsCoverImage = !material.videoCoverUrl
|
||
|
||
if (!videoSourceUrl) {
|
||
console.warn(`[TranscodeQueue] Skipping material ${materialId}: no videoSourceUrl`)
|
||
continue
|
||
}
|
||
|
||
// 重新加入队列（队列会自动调度执行）
|
||
await this.add(
|
||
materialId,
|
||
videoSourceUrl,
|
||
needsCoverImage,
|
||
taskFn
|
||
)
|
||
requeued++
|
||
} catch (error) {
|
||
console.error(`[TranscodeQueue] Failed to requeue task ${material.id}:`, error)
|
||
}
|
||
}
|
||
|
||
console.log(`[TranscodeQueue] Recovery completed: recovered ${recovered} tasks, requeued ${requeued} tasks`)
|
||
} catch (error) {
|
||
console.error('[TranscodeQueue] Recovery failed:', error)
|
||
}
|
||
|
||
return { recovered, requeued }
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 清理转码残留文件（S3上的预览视频、片段、自动生成的封面）
|
||
*
|
||
* @param material 素材信息（包含转码生成的文件URL）
|
||
*/
|
||
export async function cleanupTranscodeResidues(material: {
|
||
id: string
|
||
videoPreview540Url: string | null
|
||
videoPreviewClipUrl: string | null
|
||
videoCoverUrl: string | null
|
||
}): Promise<void> {
|
||
const { deleteFromS3, isS3PublicUrl } = await import('@/lib/s3')
|
||
|
||
const filesToDelete: string[] = []
|
||
|
||
// 1. 预览视频（540P）- 如果存在且是S3 URL，删除
|
||
if (material.videoPreview540Url && isS3PublicUrl(material.videoPreview540Url)) {
|
||
filesToDelete.push(material.videoPreview540Url)
|
||
}
|
||
|
||
// 2. 预览片段（10秒）- 如果存在且是S3 URL，删除
|
||
if (material.videoPreviewClipUrl && isS3PublicUrl(material.videoPreviewClipUrl)) {
|
||
filesToDelete.push(material.videoPreviewClipUrl)
|
||
}
|
||
|
||
// 3. 封面图 - 仅当是自动生成的（文件名包含 "video-cover-"）才删除
|
||
if (material.videoCoverUrl && isS3PublicUrl(material.videoCoverUrl)) {
|
||
const fileName = material.videoCoverUrl.split('/').pop() || ''
|
||
if (fileName.includes('video-cover-')) {
|
||
filesToDelete.push(material.videoCoverUrl)
|
||
} else {
|
||
console.log(`[TranscodeQueue] Skipping user-uploaded cover: ${material.videoCoverUrl}`)
|
||
}
|
||
}
|
||
|
||
// 批量删除文件
|
||
if (filesToDelete.length > 0) {
|
||
console.log(`[TranscodeQueue] Cleaning up ${filesToDelete.length} residue files for material ${material.id}`)
|
||
for (const url of filesToDelete) {
|
||
try {
|
||
await deleteFromS3(url)
|
||
console.log(`[TranscodeQueue] Deleted: ${url}`)
|
||
} catch (error) {
|
||
console.error(`[TranscodeQueue] Failed to delete ${url}:`, error)
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
// 单例模式
|
||
export const transcodeQueue = new VideoTranscodeQueue()
|
||
|