NestJS+Bull队列任务消失问题排查求助

核心现象解读 #

• 跨服务处理任务是Bull的正常分布式消费特性：只要两个NestJS服务连接同一个Redis实例，且使用相同的队列名称，就会组成消费者集群，共享任务池，因此3001端口提交的任务被3002端口服务处理是预期行为，并非异常。
• 任务消失则是异常情况，结合日志来看，6个任务仅2个触发完成事件，且服务触发onDrained（队列排空），说明服务认为队列已无待处理任务，但实际有4个任务未执行，可能的原因如下：

可能原因 #

1. 任务进入失败/延迟队列但未被监听捕获：你的OnQueueFailed和OnQueueError日志逻辑存在参数错误，可能导致异常被吞，无法看到任务失败的日志；部分任务可能因执行异常进入失败队列，但未触发你的监听事件。
2. Redis中任务状态异常：任务可能被意外移除、或因Redis连接波动导致状态同步失败，服务无法感知到待处理任务。
3. 并发配置限制+任务状态丢失：若每个服务的队列并发数设为1，两个服务总共可同时处理2个任务，但剩余4个任务应在这2个完成后继续执行，除非任务已从等待队列中消失。
4. Bull版本不一致：两个服务的Bull库版本不同，可能导致客户端与Redis的数据结构不兼容，引发任务丢失。

具体排查步骤 #

1. 直接检查Redis中的任务状态 #

通过Redis CLI连接到localhost:6379，执行以下命令查看队列的任务分布（替换<你的队列名称>为实际队列名）：

# 查看等待中的任务
LRANGE bull:<你的队列名称>:waiting 0 -1
# 查看活跃中的任务
LRANGE bull:<你的队列名称>:active 0 -1
# 查看失败的任务
LRANGE bull:<你的队列名称>:failed 0 -1
# 查看延迟的任务
LRANGE bull:<你的队列名称>:delayed 0 -1

• 如果这些列表中存在消失的任务ID，说明任务未丢失，只是未触发你的监听事件；如果无对应ID，说明任务已被移除。

2. 修复监听事件的日志逻辑 #

原代码中OnQueueError和OnQueueFailed的参数处理有误，导致异常日志无法正常输出，修改为：

@OnQueueFailed()
async onFailed(job: Job | undefined, error: Error) {
  console.log(`Job ${job?.id || 'unknown'} failed with error: ${error.message}`, error.stack);
}

@OnQueueError()
async onError(error: Error) {
  console.log(`Queue error occurred: ${error.message}`, error.stack);
}

重新运行服务，观察是否有错误日志输出，这是定位任务消失的关键。

3. 验证队列并发与自动移除配置 #

• 检查队列注册时的concurrency参数，确认两个服务的并发数总和是否足够处理任务：

  BullModule.registerQueue({
    name: '你的队列名称',
    redis: { host: 'localhost', port: 6379 },
    concurrency: 2, // 调整为合适值，比如每个服务设为3，总和6可一次性处理所有任务
    defaultJobOptions: {
      removeOnComplete: false, // 临时禁用任务完成后自动移除，便于排查
      removeOnFail: false, // 保留失败任务
    },
  })
  

• 重新提交任务，查看Redis中的任务状态变化，确认任务是否被正确执行或进入失败队列。

4. 检查Bull版本一致性 #

在两个服务的项目根目录下执行以下命令，确认Bull版本完全一致：

npm list bull
# 或 yarn list bull

版本不一致会导致客户端与Redis的数据结构不兼容，引发任务状态同步异常。

5. 排查Redis连接稳定性 #

• 查看EC2上的Redis日志（默认路径/var/log/redis/redis-server.log），检查是否有连接断开、超时等异常。
• 在NestJS服务中添加Redis客户端的错误监听：

  import { InjectQueue } from '@nestjs/bull';
  import { Queue } from 'bull';
  
  // 在服务的构造函数或初始化方法中
  constructor(@InjectQueue('你的队列名称') private queue: Queue) {
    this.queue.client.on('error', (err) => {
      console.log('Redis client connection error:', err);
    });
  }
  

观察是否有Redis连接异常日志。

内容的提问来源于stack exchange，提问作者Kyu