基于BullMQ与NestJS处理高并发任务的最佳实践咨询

1. 高并发场景下扩展BullMQ Worker的最佳实践 #

• 横向多实例扩展：用PM2启动多个同队列的Worker实例，比如针对通知队列启动3-5个Worker，数量可根据CPU核心数和Redis负载动态调整。PM2配置示例：

  {
    "apps": [
      {
        "name": "notification-worker",
        "script": "dist/main.js",
        "instances": "max",
        "exec_mode": "cluster",
        "env": {
          "QUEUE_NAME": "notification"
        }
      }
    ]
  }
  

• 动态调整Worker并发数：在@nestjs/bull的Worker配置中设置concurrency参数，IO密集型任务（如调用外部API发通知）可设为10-20，CPU密集型任务（如复杂可用性检查）设为CPU核心数的1-2倍。示例：

  @Processor('notification')
  export class NotificationProcessor {
    @Process({ concurrency: 15 })
    async handleNotification(job: Job) {
      // 任务逻辑
    }
  }
  

• 拆分长任务：把耗时的大任务拆成多个小任务串联执行（比如先检查可用性，再生成通知，最后同步外部服务），避免单个任务占用Worker过久导致队列阻塞。
• Redis资源优化：确保托管Redis实例的连接数足够支撑多Worker并发，开启Redis持久化避免任务丢失，同时用批量操作减少Redis命令的网络往返次数。

2. 应按任务类型还是队列优先级拆分Worker？ #

核心原则是结合任务的资源消耗特性和业务优先级：

• 优先按任务类型拆分：你的场景中，通知（IO密集、低CPU消耗、实时性要求高）和数据同步（可能涉及大量数据库读写或外部API调用、实时性要求稍低）属于不同类型任务，建议用独立的Worker集群。这样可以避免同步任务阻塞通知任务的处理。
• 高优先级任务单独分配Worker：如果某些任务（如紧急预约通知）需要优先处理，可为这类任务创建独立的高优先级队列，分配专属Worker，同时在BullMQ中设置priority参数（数值越大优先级越高）。示例：

  await this.bullService.add('notification', jobData, { priority: 10 });
  

• 低优先级任务共享Worker资源：对于非核心的同步任务，可用一个Worker集群处理多个低优先级队列，节省服务器资源。

3. 如何有效监控并避免任务阻塞/延迟？ #

• 核心指标监控：
  • 队列长度：如果队列长度持续增长，说明Worker处理能力不足，需要及时扩容。
  • 任务处理时长：统计每个任务的平均处理时间，超过阈值的任务要排查性能瓶颈。
  • 失败任务数：跟踪失败任务，设置指数退避式重试策略，将多次失败的任务转入死信队列单独处理。
• 利用BullMQ内置能力：
  • 监听Worker的completed、failed、stalled事件，实时捕获异常情况：

    @Processor('notification')
    export class NotificationProcessor {
      constructor(@InjectQueue('notification') private queue: Queue) {
        this.queue.on('stalled', (jobId) => {
          console.log(`任务 ${jobId} 已停滞，需排查`);
        });
      }
    }
    

  • 使用QueueEvents模块跟踪队列状态，或集成Prometheus暴露监控指标。
• 避免阻塞的实践：
  • 禁止在Worker中执行同步IO操作，全部用异步方法释放事件循环。
  • 设置任务超时时间：给每个任务设置timeout参数，防止任务无限期占用Worker：

    await this.bullService.add('notification', jobData, { timeout: 30000 }); // 30秒超时
    

  • 定期清理过期任务：调用clean()方法清理已完成或失败的旧任务，减少Redis内存占用。

内容的提问来源于stack exchange，提问作者Diego M. Otero