Press n or j to go to the next uncovered block, b, p or k for the previous block.
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 | /** * 队列服务 * * 【架构职责】 * 队列只负责：排队 + 并发控制 * * 【职责边界】 * - ✅ 做：任务排队、并发限制、任务分发 * - ❌ 不做：失败重试、超时管理、业务逻辑 * * 【设计原则】 * - 单一职责：队列只管排队，不管其他 * - 失败重试 → 容错层（fault-tolerance.ts） * - 超时管理 → AI服务层（llm.service.ts） * - 业务逻辑 → 领域层（langgraph-generator.ts） */ import Queue from 'bull'; import { redisService } from './redis.service'; import { memoryQueue } from './memory-queue'; // 任务队列类型 export enum QueueType { AUDIO_GENERATION = 'audio:generation', VIDEO_GENERATION = 'video:generation', BOOK_GENERATION = 'book:generation', EMAIL_SEND = 'email:send', } // 任务状态 export enum TaskStatus { WAITING = 'waiting', ACTIVE = 'active', COMPLETED = 'completed', FAILED = 'failed', DELAYED = 'delayed', } // 任务数据接口 interface TaskData { userId?: number; [key: string]: any; } // 任务进度回调 export type ProgressCallback = (progress: number, data?: any) => void; class QueueService { private queues: Map<string, Queue.Queue> = new Map(); private progressCallbacks: Map<string, ProgressCallback> = new Map(); private queueAvailable: boolean = true; constructor() { // 如果 Redis 不可用，记录警告但不阻止程序运行 if (!redisService.isAvailable()) { console.warn('[Queue] Redis 不可用，任务队列将无法正常工作'); this.queueAvailable = false; } } /** * 检查队列是否可用 */ isQueueAvailable(): boolean { return this.queueAvailable && redisService.isAvailable(); } /** * 获取或创建队列 * @param queueName 队列名称 */ private getQueue(queueName: string): Queue.Queue | null { if (!this.isQueueAvailable()) { return null; } if (!this.queues.has(queueName)) { try { const queue = new Queue(queueName, { redis: { host: process.env.REDIS_HOST || 'localhost', port: parseInt(process.env.REDIS_PORT || '6379'), password: process.env.REDIS_PASSWORD || undefined, db: parseInt(process.env.REDIS_DB || '0'), maxRetriesPerRequest: null, }, defaultJobOptions: { removeOnComplete: 100, removeOnFail: 50, }, settings: { stalledInterval: 10 * 60 * 1000, maxStalledCount: 10, } }); queue.on('completed', (job) => { console.log(`[Queue] 任务完成: ${queueName}#${job.id}`); this.clearProgressCallback(job.id.toString()); }); queue.on('failed', (job, err) => { console.error(`[Queue] 任务失败: ${queueName}#${job.id}`, err.message); this.clearProgressCallback(job.id.toString()); }); queue.on('error', (err) => { console.error(`[Queue] 队列错误: ${queueName}`, err.message); this.queueAvailable = false; }); this.queues.set(queueName, queue); console.log(`[Queue] 队列初始化: ${queueName}`); } catch (error) { console.error(`[Queue] 队列创建失败: ${queueName}`, error); this.queueAvailable = false; return null; } } return this.queues.get(queueName) || null; } /** * 添加任务到队列 * @param queueType 队列类型 * @param data 任务数据 * @param options 任务选项 * @returns 任务 ID 或 null（队列不可用时） */ async addTask( queueType: QueueType, data: TaskData, options: Queue.JobOptions = {} ): Promise<string | null> { // 如果 Redis 从未可用过，或已回退到内存队列，直接走内存队列 // 避免 Redis 变可用后任务走 Bull（无处理器）而内存队列收不到 if (!this.queueAvailable) { return this.addToMemoryQueue(queueType, data); } // 尝试 Redis 队列 const queue = this.getQueue(queueType); if (queue) { try { const job = await queue.add(data, options); console.log(`[Queue] 任务添加(Redis): ${queueType}#${job.id}`); return job.id.toString(); } catch (error) { console.error(`[Queue] Redis 任务添加失败:`, error); this.queueAvailable = false; // Fallback 到内存队列 } } // Fallback: 使用内存队列 return this.addToMemoryQueue(queueType, data); } /** 统一走内存队列 */ private async addToMemoryQueue(queueType: QueueType, data: TaskData): Promise<string | null> { console.log(`[Queue] 使用内存队列: ${queueType}`, Object.keys(data)); try { const jobId = await memoryQueue.add(queueType, data); console.log(`[Queue] 内存队列任务添加成功: ${jobId}`); return jobId; } catch (error) { console.error(`[Queue] 内存队列也失败了:`, error); return null; } } /** * 添加音频生成任务 * @param data 任务数据 */ async addAudioGenerationTask(data: TaskData): Promise<string | null> { return this.addTask(QueueType.AUDIO_GENERATION, data, { timeout: 300000, // 5 分钟超时 }); } /** * 添加视频生成任务 * @param data 任务数据 */ async addVideoGenerationTask(data: TaskData): Promise<string | null> { return this.addTask(QueueType.VIDEO_GENERATION, data, { timeout: 600000, // 10 分钟超时 }); } /** * 添加书籍生成任务 * @param data 任务数据 */ async addBookGenerationTask(data: TaskData): Promise<string | null> { return this.addTask(QueueType.BOOK_GENERATION, data, { timeout: 7200000, // 2小时超时（一本书完整生成约40-60分钟） }); } /** * 获取任务状态 * @param queueType 队列类型 * @param jobId 任务 ID */ async getTaskStatus(queueType: QueueType, jobId: string): Promise<{ status: TaskStatus; progress: number; data?: any; error?: string; }> { const queue = this.getQueue(queueType); if (!queue) { return { status: TaskStatus.FAILED, progress: 0, error: '队列不可用', }; } const job = await queue.getJob(jobId); if (!job) { return { status: TaskStatus.FAILED, progress: 0, error: '任务不存在', }; } const state = await job.getState(); const statusMap: Record<string, TaskStatus> = { waiting: TaskStatus.WAITING, active: TaskStatus.ACTIVE, completed: TaskStatus.COMPLETED, failed: TaskStatus.FAILED, delayed: TaskStatus.DELAYED, }; return { status: statusMap[state] || TaskStatus.FAILED, progress: job.progress() || 0, data: job.returnvalue, error: job.failedReason, }; } /** * 更新任务进度 * @param queueType 队列类型 * @param jobId 任务 ID * @param progress 进度 (0-100) * @param data 附加数据 */ async updateProgress( queueType: QueueType, jobId: string, progress: number, data?: any ): Promise<void> { const queue = this.getQueue(queueType); if (!queue) { return; // 队列不可用时静默忽略 } const job = await queue.getJob(jobId); if (job) { await job.progress(progress); // 触发进度回调 const callback = this.progressCallbacks.get(jobId); if (callback) { callback(progress, data); } } } /** * 注册进度回调 * @param jobId 任务 ID * @param callback 回调函数 */ onProgress(jobId: string, callback: ProgressCallback): void { this.progressCallbacks.set(jobId, callback); } /** * 清除进度回调 * @param jobId 任务 ID */ clearProgressCallback(jobId: string): void { this.progressCallbacks.delete(jobId); } /** * 获取队列统计信息 * @param queueType 队列类型 */ async getQueueStats(queueType: QueueType): Promise<{ waiting: number; active: number; completed: number; failed: number; delayed: number; }> { const queue = this.getQueue(queueType); const counts = await queue.getJobCounts(); return counts; } /** * 清空队列 * @param queueType 队列类型 */ async clearQueue(queueType: QueueType): Promise<void> { const queue = this.getQueue(queueType); await queue.empty(); console.log(`[Queue] 队列清空: ${queueType}`); } /** * 暂停队列 * @param queueType 队列类型 */ async pauseQueue(queueType: QueueType): Promise<void> { const queue = this.getQueue(queueType); await queue.pause(); console.log(`[Queue] 队列暂停: ${queueType}`); } /** * 恢复队列 * @param queueType 队列类型 */ async resumeQueue(queueType: QueueType): Promise<void> { const queue = this.getQueue(queueType); await queue.resume(); console.log(`[Queue] 队列恢复: ${queueType}`); } /** * 关闭所有队列 */ async closeAll(): Promise<void> { for (const [name, queue] of this.queues) { await queue.close(); console.log(`[Queue] 队列关闭: ${name}`); } this.queues.clear(); } } // 导出单例 export const queueService = new QueueService(); export default queueService; |