mcpField
N
nestjs-queue-architect

nestjs-queue-architect

github.com/shipshitdev开发38311 次安装2 天前更新
代码仓库

关于 nestjs-queue-architect

面向开发者的架构技能，用于构建 NestJS 与 BullMQ 作业处理系统，尤其适合媒体密集型工作流。内容涵盖队列服务、处理器、重试、错误处理和清理模式。

主要功能

  1. 01组织队列服务与处理器
  2. 02按作业名称路由
  3. 03配置重试与退避策略
  4. 04处理错误并清理临时文件
  5. 05实现幂等作业处理器

适用场景

  1. 01设计媒体处理队列
  2. 02实现作业重试机制
  3. 03组织 NestJS 作业路由
  4. 04避免处理过程遗留临时文件

安装

复制下面的命令，在终端中运行。

$npx skills add https://github.com/shipshitdev/skills --skill nestjs-queue-architect

安装后在 Claude Code 中输入/nestjs-queue-architect确认已加载。

附带文件plugin.jsonreferences/full-guide.md

详细介绍

NestJS 队列架构师 - BullMQ 专家

为媒体密集型工作流设计弹性、可扩展的 BullMQ + NestJS 作业处理系统。

技术栈

  • BullMQ：5.61.0（基于 Redis 的作业队列）
  • @nestjs/bullmq：12.0.0（NestJS 12）
  • @bull-board/nestjs：6.13.1（队列监控界面）

项目上下文探索

开始实现前：

  1. 检查 .agents/memory/ 中是否有描述队列模式的架构文件
  2. 查看现有队列服务和常量
  3. 查找 [project]-queue-architect 技能

核心模式

队列常量
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;
队列服务
@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 },
    });
  }
}
处理器（WorkerHost）
@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}`);
    }
  }
}

关键原则

  1. 每种队列类型对应一个服务 - 封装作业选项
  2. 基于 switch 的路由 - 按 job.name 路由
  3. 结构化错误处理 - 记录日志、发送 WebSocket、发布 Redis 消息，然后重新抛出错误
  4. 始终清理 - 在 try/finally 中处理临时文件
  5. 幂等处理器 - 可安全重试

队列配置

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

重试策略

作业类型尝试次数延迟原因
调整大小32000ms暂时性故障
合并25000ms资源密集型
元数据21000ms速度快，快速失败
清理51000ms必须成功

常见问题

  • 内存泄漏：始终设置 removeOnComplete/Fail
  • 超时：为繁重作业设置合适的 timeout
  • 竞态条件：确保处理器具备幂等性

完整的处理器示例、测试模式、Bull Board 配置和 Redis 发布/订阅集成，请参阅： references/full-guide.md

评论

支持 Markdown