定时任务与任务调度

环境信息：Node.js 20.x、@nestjs/schedule 4.x

安装配置#

定时任务在后端开发中很常见，比如定时清理日志、定时发送报告、定时同步数据等。

NestJS 提供了 @nestjs/schedule 模块来处理定时任务：

pnpm add @nestjs/schedule -S

在 AppModule 中引入：

import { ScheduleModule } from '@nestjs/schedule'

@Module({
  imports: [ScheduleModule.forRoot()],
})
export class AppModule {}

Cron 任务#

使用 @Cron() 装饰器定义定时任务：

import { Injectable } from '@nestjs/common'
import { Cron, CronExpression } from '@nestjs/schedule'

@Injectable()
export class TaskService {
  @Cron('0 0 8 * * *')
  handleMorningTask() {
    console.log('每天早上 8 点执行')
  }

  @Cron(CronExpression.EVERY_30_MINUTES)
  handleHalfHourTask() {
    console.log('每 30 分钟执行一次')
  }
}

Cron 表达式#

Cron 表达式由 6 个字段组成：

┌──────────── 秒 (0-59)
│ ┌────────── 分钟 (0-59)
│ │ ┌──────── 小时 (0-23)
│ │ │ ┌────── 日 (1-31)
│ │ │ │ ┌──── 月 (1-12)
│ │ │ │ │ ┌── 星期 (0-6，0 表示周日)
│ │ │ │ │ │
* * * * * *

常用示例：

'0 0 8 * * *' // 每天 8:00
'0 30 9 * * 1-5' // 周一到周五 9:30
'0 0 */2 * * *' // 每 2 小时
'0 0 0 1 * *' // 每月 1 号 0:00

NestJS 也提供了预定义的表达式：

import { CronExpression } from '@nestjs/schedule'

CronExpression.EVERY_MINUTE // 每分钟
CronExpression.EVERY_HOUR // 每小时
CronExpression.EVERY_DAY_AT_MIDNIGHT // 每天 0:00
CronExpression.EVERY_WEEK // 每周
CronExpression.EVERY_WEEKDAY // 每个工作日

Interval 任务#

按固定间隔执行任务：

import { Interval } from '@nestjs/schedule'

@Injectable()
export class TaskService {
  @Interval(5000)
  handleInterval() {
    console.log('每 5 秒执行一次')
  }

  @Interval('cleanup-task', 60000)
  handleCleanup() {
    console.log('每分钟清理一次')
  }
}

Timeout 任务#

延迟执行一次：

import { Timeout } from '@nestjs/schedule'

@Injectable()
export class TaskService {
  @Timeout(10000)
  handleTimeout() {
    console.log('应用启动 10 秒后执行一次')
  }
}

动态任务管理#

可以在运行时动态添加、暂停、删除任务：

import { Injectable } from '@nestjs/common'
import { SchedulerRegistry } from '@nestjs/schedule'
import { CronJob } from 'cron'

@Injectable()
export class TaskService {
  constructor(private schedulerRegistry: SchedulerRegistry) {}

  // 动态添加 Cron 任务
  addCronJob(name: string, cronTime: string) {
    const job = new CronJob(cronTime, () => {
      console.log(`任务 ${name} 执行`)
    })

    this.schedulerRegistry.addCronJob(name, job)
    job.start()
    console.log(`任务 ${name} 已添加`)
  }

  // 删除任务
  deleteCronJob(name: string) {
    this.schedulerRegistry.deleteCronJob(name)
    console.log(`任务 ${name} 已删除`)
  }

  // 暂停任务
  stopCronJob(name: string) {
    const job = this.schedulerRegistry.getCronJob(name)
    job.stop()
    console.log(`任务 ${name} 已暂停`)
  }

  // 恢复任务
  startCronJob(name: string) {
    const job = this.schedulerRegistry.getCronJob(name)
    job.start()
    console.log(`任务 ${name} 已恢复`)
  }

  // 获取所有任务
  getCronJobs() {
    const jobs = this.schedulerRegistry.getCronJobs()
    jobs.forEach((value, key) => {
      const next = value.nextDate().toJSDate()
      console.log(`任务: ${key}, 下次执行: ${next}`)
    })
  }
}

任务锁#

在分布式环境中，多个实例可能同时执行同一个定时任务。可以使用 Redis 实现分布式锁：

import { Injectable } from '@nestjs/common'
import { Cron } from '@nestjs/schedule'
import { Redis } from 'ioredis'

@Injectable()
export class TaskService {
  private redis = new Redis()
  private lockKey = 'task:report:lock'
  private lockTTL = 60 // 锁过期时间 60 秒

  @Cron('0 0 9 * * *')
  async handleDailyReport() {
    // 尝试获取锁
    const acquired = await this.redis.set(
      this.lockKey,
      '1',
      'EX',
      this.lockTTL,
      'NX'
    )

    if (!acquired) {
      console.log('其他实例正在执行任务')
      return
    }

    try {
      console.log('执行每日报告任务')
      // 执行任务逻辑
    } finally {
      // 释放锁
      await this.redis.del(this.lockKey)
    }
  }
}

任务执行日志#

记录任务执行情况：

import { Injectable, Logger } from '@nestjs/common'
import { Cron } from '@nestjs/schedule'

@Injectable()
export class TaskService {
  private readonly logger = new Logger(TaskService.name)

  @Cron('0 */10 * * * *')
  async handleDataSync() {
    const startTime = Date.now()
    this.logger.log('数据同步任务开始')

    try {
      // 执行同步逻辑
      await this.syncData()

      const duration = Date.now() - startTime
      this.logger.log(`数据同步完成，耗时 ${duration}ms`)
    } catch (error) {
      this.logger.error('数据同步失败', error.stack)
    }
  }

  private async syncData() {
    // 同步逻辑
  }
}