Node.js中动态截止日期的事件调度与变更处理方案咨询

数据库中存储着可动态修改的截止日期：

deadline A: 11:30am
deadline B: 4:50pm
deadline C: 6:15pm

需要在每个截止日期的前120分钟、前30分钟、截止时刻分别发送对应通知，示例如下：

9:30am:  "upcoming deadline A!"
11:00am: "upcoming deadline A!"
11:30am: "deadline passed A!"
2:50pm:  "upcoming deadline B!"
4:15pm:  "upcoming deadline C!"
4:20pm:  "upcoming deadline B!"
4:50pm:  "deadline passed B!"
5:45pm:  "upcoming deadline C!"
6:15pm:  "deadline passed C!"

一、核心设计思路 #

采用定期轮询+即时事件触发的组合模式，避免一次性调度所有未来任务的局限性：

• 定期轮询数据库，检查所有截止日期的三个触发时间点，补全未调度的任务
• 截止日期变更时，立即触发任务重调度，取消旧任务并添加新任务

二、具体实现步骤 #

1. 选择持久化调度库 #

优先选支持任务持久化的库，比如bullmq——依托Redis实现任务持久化，应用重启后未执行的任务不会丢失，适合长期运行场景。

2. 定义任务规则与唯一标识 #

每个截止日期对应三个触发时间：

• 触发时间1：截止时间 - 120分钟
• 触发时间2：截止时间 - 30分钟
• 触发时间3：截止时刻

为每个任务生成唯一标识，格式如deadline-{ID}-{type}，其中type为120min/30min/due，方便后续修改时精准定位并删除旧任务。

3. 初始化与轮询调度 #

• 应用启动时，遍历数据库所有截止日期，计算三个触发时间，将未过期的任务加入调度队列
• 启动定时轮询（比如每分钟执行一次），检查新添加的截止日期或未被调度的触发时间，补充任务到队列

4. 任务执行逻辑 #

任务触发时执行消息发送，同时记录执行状态到数据库，避免重复发送。示例代码片段：

const { Queue, Worker } = require('bullmq');

// 初始化通知队列
const deadlineQueue = new Queue('deadline-notifications');

// 任务处理器：执行消息发送
const worker = new Worker('deadline-notifications', async (job) => {
  const { deadlineId, message } = job.data;
  // 替换为实际消息发送逻辑（如推送通知、调用API）
  console.log(`${new Date().toLocaleTimeString()}: "${message}"`);
  // 记录任务执行状态到数据库，避免重复执行
  await markTaskAsCompleted(deadlineId, job.name);
});

// 计算三个触发时间点
function calculateTriggerTimes(deadline) {
  const deadlineDate = new Date(deadline);
  return [
    new Date(deadlineDate.getTime() - 120 * 60 * 1000), // 提前120分钟
    new Date(deadlineDate.getTime() - 30 * 60 * 1000),  // 提前30分钟
    deadlineDate // 截止时刻
  ];
}

// 调度单个截止日期的所有通知任务
async function scheduleDeadlineTasks(deadlineId, deadline) {
  // 先删除该截止日期的所有旧任务，避免重复调度
  await deadlineQueue.removeJobs(`deadline-${deadlineId}-*`);

  const triggerTimes = calculateTriggerTimes(deadline);
  const messages = [
    `upcoming deadline ${deadlineId}!`,
    `upcoming deadline ${deadlineId}!`,
    `deadline passed ${deadlineId}!`
  ];

  for (let i = 0; i < triggerTimes.length; i++) {
    const triggerTime = triggerTimes[i];
    // 只调度未来的任务
    if (triggerTime > new Date()) {
      await deadlineQueue.add(
        `deadline-${deadlineId}-${i === 0 ? '120min' : i === 1 ? '30min' : 'due'}`,
        { deadlineId, message: messages[i] },
        { delay: triggerTime.getTime() - Date.now() }
      );
    }
  }
}

三、处理截止日期变更 #

1. 监听数据库变更：利用数据库的变更监听能力（如MongoDB Change Stream、PostgreSQL LISTEN/NOTIFY），当截止日期被修改时，立即调用scheduleDeadlineTasks重调度任务
2. 轮询兜底：即使变更监听失效，定期轮询会检查到截止日期变化，补全或更新调度，避免遗漏
3. 幂等性保障：每次重调度前先删除该截止日期的所有旧任务，确保不会同时存在新旧任务导致重复通知

四、可靠性强化措施 #

• 任务持久化：依赖bullmq的Redis持久化，应用重启后未执行的任务自动恢复
• 重试机制：给任务配置重试策略（如失败后重试3次），应对网络波动等临时问题
• 状态记录：将任务的调度/执行/失败状态存入数据库，方便排查问题与去重
• 监控告警：配置队列监控，当任务失败次数过多或队列积压时触发告警，及时介入处理

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