BullMQ中实现单实例可重复任务的配置方法

背景介绍

在使用消息队列系统BullMQ时，开发者经常会遇到需要配置周期性重复执行任务的需求。一个常见的场景是：我们希望任务每隔固定时间运行一次，但同时要确保任何时候都只有一个任务实例在执行。当任务执行时间超过设定的间隔时，系统应该自动跳过下一次调度，等待当前任务完成后再继续。

问题分析

在BullMQ中，通过repeat选项可以轻松设置周期性任务。但默认情况下，如果任务执行时间超过间隔时间，系统会继续创建新的任务实例，这可能导致多个相同任务同时运行，造成资源竞争或数据不一致的问题。

解决方案

BullMQ的最新版本提供了更清晰的重复策略API，其中默认行为已经符合单实例运行的需求。具体来说：

1. 默认重复策略：当使用repeat选项创建周期性任务时，如果前一个任务实例仍在运行，系统会自动跳过创建新的实例，直到当前任务完成。

2. 配置示例：

await queue.add('testjob', {}, {
  repeat: { 
    every: 120000 // 每2分钟执行一次
  }
});

3. 并发控制：通过设置worker的并发数可以进一步控制任务执行：

const worker = new Worker(queueName, processor, {
  concurrency: 1 // 确保同一时间只有一个worker处理任务
});

高级配置建议

1. 任务去重：虽然默认行为已经满足需求，但可以使用jobId或debounce选项来确保任务唯一性：

await queue.add('testjob', {}, {
  jobId: 'unique-job-id',
  repeat: { every: 120000 }
});

2. 任务清理：对于频繁执行的周期性任务，建议设置removeOnComplete选项自动清理已完成任务，避免队列膨胀：

await queue.add('testjob', {}, {
  repeat: { every: 120000 },
  removeOnComplete: 100 // 保留最近100个已完成任务
});

3. 错误处理：为worker添加错误处理逻辑，确保任务失败不会影响后续调度：

worker.on('failed', (job, err) => {
  console.error(`Job ${job.id} failed:`, err);
});

最佳实践

1. 合理设置间隔时间：确保重复间隔大于任务的平均执行时间，避免频繁跳过执行。

2. 监控任务执行时间：记录任务执行耗时，为调整间隔时间提供依据。

3. 考虑使用cron表达式：对于需要复杂调度规则的任务，可以使用cron模式：

repeat: { pattern: '*/2 * * * *' } // 每2分钟执行

4. 测试不同负载场景：在生产环境部署前，模拟高负载情况验证任务调度行为是否符合预期。

通过合理配置BullMQ的重复任务选项和worker参数，开发者可以轻松实现"单实例周期性任务"的需求，确保系统稳定可靠地运行。