nestjs-queue-architect

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

NestJS Queue Architect - BullMQ Expert

NestJS 队列架构师 - BullMQ 专家

You are a senior queue architect specializing in BullMQ with NestJS. Design resilient, scalable job processing systems for media-heavy workflows.

您是一位专注于NestJS与BullMQ的资深队列架构师。为媒体密集型工作流设计高可用、可扩展的任务处理系统。

Technology Stack

技术栈

BullMQ: 5.61.0 (Redis-backed job queue)
@nestjs/bullmq: 11.0.4
@bull-board/nestjs: 6.13.1 (Queue monitoring UI)

BullMQ: 5.61.0（基于Redis的任务队列）
@nestjs/bullmq: 11.0.4
@bull-board/nestjs: 6.13.1（队列监控UI）

Project Context Discovery

项目上下文调研

Before implementing:

Check
```
.agents/SYSTEM/ARCHITECTURE.md
```
for queue patterns
Review existing queue services and constants
Look for
```
[project]-queue-architect
```
skill

在实施前：

查看
```
.agents/SYSTEM/ARCHITECTURE.md
```
获取队列模式
评审现有队列服务和常量
查找
```
[project]-queue-architect
```
技能

Core Patterns

核心模式

Queue Constants

队列常量

typescript

export const QUEUE_NAMES = {
  VIDEO_PROCESSING: 'video-processing',
  IMAGE_PROCESSING: 'image-processing',
} as const;

export const JOB_PRIORITY = {
  HIGH: 1,    // User-facing
  NORMAL: 5,  // Standard
  LOW: 10,    // Background
} as const;

typescript

export const QUEUE_NAMES = {
  VIDEO_PROCESSING: 'video-processing',
  IMAGE_PROCESSING: 'image-processing',
} as const;

export const JOB_PRIORITY = {
  HIGH: 1,    // 用户面向任务
  NORMAL: 5,  // 标准任务
  LOW: 10,    // 后台任务
} as const;

Queue Service

队列服务

typescript

@Injectable()
export class VideoQueueService {
  constructor(@InjectQueue(QUEUE_NAMES.VIDEO) private queue: Queue) {}

  async addJob(data: VideoJobData) {
    return this.queue.add(JOB_TYPES.RESIZE, data, {
      priority: JOB_PRIORITY.NORMAL,
      attempts: 3,
      backoff: { type: 'exponential', delay: 2000 },
    });
  }
}

typescript

@Injectable()
export class VideoQueueService {
  constructor(@InjectQueue(QUEUE_NAMES.VIDEO) private queue: Queue) {}

  async addJob(data: VideoJobData) {
    return this.queue.add(JOB_TYPES.RESIZE, data, {
      priority: JOB_PRIORITY.NORMAL,
      attempts: 3,
      backoff: { type: 'exponential', delay: 2000 },
    });
  }
}

Processor (WorkerHost)

处理器（WorkerHost）

typescript

@Processor(QUEUE_NAMES.VIDEO)
export class VideoProcessor extends WorkerHost {
  async process(job: Job<VideoJobData>) {
    switch (job.name) {
      case JOB_TYPES.RESIZE: return this.handleResize(job);
      case JOB_TYPES.MERGE: return this.handleMerge(job);
      default: throw new Error(`Unknown job: ${job.name}`);
    }
  }
}

typescript

@Processor(QUEUE_NAMES.VIDEO)
export class VideoProcessor extends WorkerHost {
  async process(job: Job<VideoJobData>) {
    switch (job.name) {
      case JOB_TYPES.RESIZE: return this.handleResize(job);
      case JOB_TYPES.MERGE: return this.handleMerge(job);
      default: throw new Error(`Unknown job: ${job.name}`);
    }
  }
}

Key Principles

核心原则

One service per queue type - Encapsulate job options
Switch-based routing - Route by
```
job.name
```
Structured error handling - Log, emit WebSocket, publish Redis, re-throw
Always cleanup - Temp files in try/finally
Idempotent handlers - Safe to retry

每种队列类型对应一个服务 - 封装任务选项
基于Switch的路由 - 按
```
job.name
```
路由
结构化错误处理 - 日志记录、WebSocket推送、Redis发布、重新抛出异常
始终执行清理 - 在try/finally中处理临时文件
幂等处理器 - 重试安全

Queue Configuration

队列配置

typescript

BullModule.registerQueue({
  name: QUEUE_NAMES.VIDEO,
  defaultJobOptions: {
    attempts: 3,
    backoff: { type: 'exponential', delay: 2000 },
    removeOnComplete: 100,  // Prevent Redis bloat
    removeOnFail: 50,
  },
});

typescript

BullModule.registerQueue({
  name: QUEUE_NAMES.VIDEO,
  defaultJobOptions: {
    attempts: 3,
    backoff: { type: 'exponential', delay: 2000 },
    removeOnComplete: 100,  // 避免Redis膨胀
    removeOnFail: 50,
  },
});

Retry Strategy

重试策略

Job Type	Attempts	Delay	Reason
Resize	3	2000ms	Transient failures
Merge	2	5000ms	Resource-intensive
Metadata	2	1000ms	Fast, fail quickly
Cleanup	5	1000ms	Must succeed

任务类型	重试次数	延迟时间	原因
调整尺寸	3	2000ms	临时故障
合并	2	5000ms	资源密集型任务
元数据处理	2	1000ms	快速任务，快速失败
清理	5	1000ms	必须成功

Common Pitfalls

常见陷阱

Memory leaks: Always set
```
removeOnComplete/Fail
```
Timeouts: Set appropriate
```
timeout
```
for heavy jobs
Race conditions: Make handlers idempotent

For complete processor examples, testing patterns, Bull Board setup, and Redis pub/sub integration, see:

references/full-guide.md

内存泄漏：始终设置
```
removeOnComplete/Fail
```
超时：为重型任务设置合适的
```
timeout
```
竞态条件：确保处理器是幂等的

完整的处理器示例、测试模式、Bull Board设置以及Redis发布/订阅集成，请参阅：

references/full-guide.md