Node.js如何通过数组批量执行determineCron替代硬编码逐个调用

优化点说明 #

原有实现存在两个明显的维护问题：

• 每条DAG运行记录单独声明常量，新增处理对象时需要重复编写常量定义、函数调用代码，冗余度高
• 函数内变量未声明、SQL直接字符串拼接存在全局污染、注入风险，且单条任务报错会直接中断整个流程

优化后统一将所有待处理任务存入数组管理，通过遍历自动完成所有任务的Cron解析，无需硬编码单个调用。

优化后完整代码 #

// 待处理任务统一存入数组，新增任务直接往数组追加对象即可
const taskRuns = [
  {
    start_date: '2022-07-01T13:45:01.221636+00:00',
    end_date: '2022-07-01T14:11:01.864293+00:00',
    state: 'success',
    dag_run_id: 'scheduled__2022-06-30T13:45:00+00:00',
    dag_id: 'IPP_CYCLE_PARMS',
  },
  {
    start_date: '2022-06-26T23:00:00.742495+00:00',
    end_date: '2022-06-27T14:10:23.108401+00:00',
    state: 'failed',
    dag_run_id: 'scheduled__2022-06-25T23:00:00+00:00',
    dag_id: 'EFS-Winning-Route-daily-batch'
  }
]

async function determineCron(result){
  const dagID = result?.dag_id || 0
  if (!dagID) return
  console.log('当前处理DAG ID:', dagID)
  // 参数化查询避免SQL注入，可根据实际使用的BigQuery客户端调整参数写法
  const job = await bqConnection().query({
    query: `SELECT * FROM \`np-inventory-planning-thd.IPP_SLA.expected_sla\` where dag_id = @dagId`,
    params: { dagId: dagID }
  })
  const slaConfig = job[0][0]
  if (!slaConfig?.cron_time) {
    console.log('未匹配到对应Cron配置，DAG ID:', dagID)
    return
  }
  const interval = parser.parseExpression(slaConfig.cron_time);
  console.log('下次执行时间: ', interval.next().toString());
}

// 批量处理入口
async function batchProcessTasks() {
  // 串行处理：逐个执行任务，不会占用过多数据库连接，适合任务量较大的场景
  for (const run of taskRuns) {
    try {
      await determineCron(run)
    } catch (err) {
      console.error('任务处理失败，DAG ID:', run.dag_id, err)
    }
  }

  // 如果需要并行处理提升速度，注释掉上面的for循环，放开下面这段代码即可
  // await Promise.all(taskRuns.map(run => determineCron(run).catch(err => {
  //   console.error('任务处理失败，DAG ID:', run.dag_id, err)
  // })))
}

// 启动批量处理
batchProcessTasks()

使用注意 #

• 后续新增待解析任务，只需要把对应run对象追加到taskRuns数组中，不需要额外编写determineCron()调用语句
• 两种执行模式按需选择：串行模式对数据库压力小，并行模式处理速度快
• 增加了单任务错误捕获、空值校验逻辑，单条任务异常不会中断整个批量流程

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