Node.js中Bull队列作业2.5小时后重入队引发重复处理问题排查

问题背景 #

在Node.js环境中基于Bull队列实现大文件拆分后的数据库插入流程：将大文件拆分为单个页面数据后，按CHUNK_SIZE=30分块生成约736个作业推入队列。

生产环境出现异常：3个作业被重复处理，导致数据库出现重复记录。原记录于凌晨1:12插入，重复记录于凌晨3:40出现，间隔约2.5小时；且重复记录无元数据，为空记录。该问题仅在生产环境出现，UAT及本地环境可正常处理736条记录无重复。

队列配置与代码 #

生产者代码（拆分作业并推入队列） #

// Producer — splits jobs into chunks and pushes to queue
const jobChunks = _.chunk(allJobs, parseInt(process.env.CHUNK_SIZE)); // CHUNK_SIZE = 30

for (let chunk of jobChunks) {
    await insertQueue.add({ jobs: chunk }, { 
        removeOnComplete: true 
    });
}

队列配置 #

// Queue configuration
const queue = new Bull('insert-queue', REDIS_URL, {
    settings: {
        maxStalledCount: 1,
        stalledInterval: 30000,
    },
    redis: {
        connectTimeout: 10000, // 10 seconds
    },
    defaultJobOptions: {
        timeout: 300000, // 5 minutes
    },
});

消费者代码（处理分块作业） #

// Consumer — processes each chunk
queue.process(1, async (job, done) => {
    try {
        await insertRecordsInDb(job.data.jobs); // inserts 30 records per job
        done();
    } catch (error) {
        done(error);
    }
});

环境参数 #

• CHUNK_SIZE=30
• REDIS_CONNECTION_TIMEOUT=10000
• INSERT_CONCURRENT_PROCESS=1
• INSERT_MAX_QUEUE=2

核心疑问 #

• 作业重入队的原因是什么？
• 若为作业停滞，停滞原因及2.5小时间隔的缘由是什么？

问题分析与解答 #

1. 作业重入队的直接原因：Bull的停滞作业重试机制 #

从配置可见，你开启了Bull的停滞作业检测：maxStalledCount:1表示作业被标记为停滞1次后就会重新入队，stalledInterval:30000表示每30秒检查一次停滞作业。当Bull判定某个作业处于停滞状态时，会自动将其重新加入队列执行，这就是重复处理的直接触发点。

2. 作业停滞的可能诱因（生产环境特有） #

结合重复记录为空的现象，大概率是生产环境特有的资源或状态异常：

• Redis连接不稳定：生产环境Redis可能出现短暂断连、主从切换或集群延迟，导致消费者进程无法向Redis发送心跳包更新作业状态。Bull长时间收不到作业的状态反馈，就会判定作业停滞。
• 数据库操作阻塞/无响应：凌晨时段可能存在数据库备份、索引重建等运维操作，导致insertRecordsInDb执行超时甚至进程挂起，但未抛出明确异常（比如连接池卡住但未触发超时错误），消费者进程无法完成done()回调，作业状态一直停留在"processing"，最终被判定为停滞。
• 进程资源瓶颈：生产环境处理大文件后，进程内存占用过高或CPU负载突增，导致进程短暂卡顿，无法按时向Redis发送状态更新，触发停滞判定。

3. 2.5小时间隔的本质 #

这个间隔并非Bull的默认重试间隔，而是作业从开始处理到被判定为停滞，再到重新调度执行的总时长，可能的场景组合：

• 第一次处理时，数据库操作卡住长达2小时，之后进程恢复响应，Bull检测到作业已停滞，重新入队后加上队列中剩余作业的等待时间，最终在2.5小时后执行重复作业。
• 生产环境Redis出现状态同步延迟，作业的"processing"状态未及时更新，直到2.5小时后Bull才完成状态校验，触发停滞重入队逻辑。

针对性修复建议 #

• 优化消费者回调逻辑：改用Bull支持的Promise风格返回，避免done回调丢失的问题：

  queue.process(1, async (job) => {
      await insertRecordsInDb(job.data.jobs);
  });
  

• 增加作业数据校验：在数据库插入前校验数据完整性，避免插入空记录：

  async function insertRecordsInDb(jobs) {
      if (!Array.isArray(jobs) || jobs.length === 0) {
          throw new Error('Invalid job data: no records to insert');
      }
      // 原数据库插入逻辑
  }
  

• 调整停滞检测与重试配置：适配生产环境实际性能，减少误判并限制重试次数：

  settings: {
      maxStalledCount: 2, // 增加停滞判定次数，降低误判概率
      stalledInterval: 60000, // 延长停滞检测间隔至1分钟
  },
  defaultJobOptions: {
      timeout: 600000, // 调大作业超时时间至10分钟
      attempts: 2, // 限制最大重试次数，避免多次重复插入
      backoff: { type: 'exponential', delay: 60000 } // 指数退避重试，避免瞬间重复执行
  }
  

• 增加监控告警：在生产环境监控Redis连接状态、作业处理时长、数据库性能指标，及时定位周期性异常。

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