周期任务与 Job Schedulers

周期性清理、日报生成和定时同步不能靠 setInterval 稳定完成：进程重启后计时器会丢失，多实例又可能重复执行。BullMQ 的 Job Scheduler 将调度配置保存在 Redis 中，并按规则持续产生普通任务。

创建或更新调度器

当前 API 是 queue.upsertJobScheduler()。upsert 表示：ID 不存在时创建，已存在时更新，适合应用重复部署而不制造重复配置。

import { Queue } from "bullmq";

interface CleanupJobData {
  scope: "temporary-files" | "expired-sessions";
}

const maintenanceQueue = new Queue<CleanupJobData>("maintenance", {
  connection: { host: "127.0.0.1", port: 6379 },
});

await maintenanceQueue.upsertJobScheduler(
  "cleanup-temporary-files",
  { every: 60_000 },
  {
    name: "cleanup",
    data: { scope: "temporary-files" },
    opts: {
      attempts: 3,
      backoff: { type: "exponential", delay: 1_000 },
      removeOnComplete: 100,
      removeOnFail: 500,
    },
  },
);

第三个参数是任务模板，定义每次生成任务的名称、数据和选项。再次以相同 scheduler ID 调用即可更新规则或模板。

every：固定间隔

await maintenanceQueue.upsertJobScheduler(
  "session-cleanup",
  {
    every: 5 * 60_000,
    limit: 100,
  },
  {
    name: "cleanup",
    data: { scope: "expired-sessions" },
  },
);

every 使用毫秒。固定间隔会对齐到时间边界。当前 BullMQ 对新创建的 scheduler 会立即产生第一次执行；immediately 从 5.19.0 起已弃用，因为当前默认行为等同于首次立即执行。更新已存在的 scheduler 不会额外立即触发一次。

pattern：Cron 表达式

BullMQ 的 cron 模式支持可选的“秒”字段：

await maintenanceQueue.upsertJobScheduler(
  "daily-cleanup",
  {
    pattern: "0 15 3 * * *",
    tz: "Asia/Shanghai",
  },
  {
    name: "cleanup",
    data: { scope: "temporary-files" },
  },
);

上例表示每天 03:15:00 执行。显式配置 tz，不要依赖服务器本地时区。需要 UTC 时使用 tz: "UTC"。

常用通用选项还包括：

startDate：从指定时间起生效。
endDate：到指定时间后停止产生任务。
limit：最多产生多少次。

调度频率不等于保证执行频率

Job Scheduler 在上一份任务开始处理时产生下一份任务。如果没有 Worker、并发不足或队列非常繁忙，任务不会为了追赶时间而无限堆积，实际处理间隔可能大于配置间隔。

这意味着：

它适合“周期执行一项工作”。
它不保证在每个时间点都生成一份永不缺失的历史记录。
业务若要求每个结算周期绝不能漏，应在任务数据中计算和记录业务周期，并由数据库约束防重、补漏。

查询和删除调度器

const scheduler = await maintenanceQueue.getJobScheduler(
  "cleanup-temporary-files",
);

const firstTen = await maintenanceQueue.getJobSchedulers(0, 9, true);

const removed = await maintenanceQueue.removeJobScheduler(
  "cleanup-temporary-files",
);

console.log({ scheduler, firstTen, removed });

getJobScheduler(id) 查询单个配置。
getJobSchedulers(start, end, asc) 分页查询配置。
removeJobScheduler(id) 返回是否确实删除了配置。

周期任务会使用特殊任务 ID，不能为生成的每个任务指定自定义 jobId。需要区分任务时使用 scheduler ID、任务 name 和业务数据。

从旧 Repeatable API 演进而来

旧教程常见以下写法：

// 旧 API，仅用于识别历史代码
await maintenanceQueue.add(
  "cleanup",
  { scope: "temporary-files" },
  { repeat: { every: 60_000 } },
);

官方说明：旧 Repeatable APIs 从 BullMQ 5.16.0 起弃用，并在 v6 移除，改用 Job Schedulers。旧的 repeat、getRepeatableJobs()、removeRepeatable() 等示例只适合历史版本。新项目应使用：

upsertJobScheduler()
getJobScheduler() / getJobSchedulers()
removeJobScheduler()

升级现有项目时不要只改方法名，应按官方 v5 到 v6 迁移指南检查已保存的旧调度数据。

实用技巧与最佳实践

使用稳定、可读的 scheduler ID，例如 billing:monthly:tenant-42。
部署时执行幂等的 upsert，不要每个请求都创建调度器。
Cron 一律显式指定 tz，并在日志中记录业务时区。
Worker 仍须幂等；周期任务可能重试或因停滞而重复执行。
在数据库为“任务类型 + 业务周期”建立唯一约束，避免月报重复生成。
监控调度器的下次执行时间、任务失败率和队列延迟。
定期清理已废弃租户或功能对应的 scheduler，避免“幽灵任务”。

本章小结

Job Scheduler 是持久化的“任务工厂”，every 适合固定间隔，pattern 适合日历时间。使用稳定 ID 配合 upsertJobScheduler() 可以安全部署和更新；同时要理解调度频率受 Worker 容量影响，不能替代业务上的补漏与幂等设计。