如何在PM2集群模式下让@nestjs/schedule仅启动单个定时任务？

方案1：仅在主进程启动定时任务 #

PM2集群模式下每个工作进程都会初始化NestJS应用，导致定时任务重复注册。可通过判断当前进程实例ID，只在指定进程中启用定时任务。

• 在ScheduleModule初始化时添加条件判断：

import { Module } from '@nestjs/common';
import { ScheduleModule } from '@nestjs/schedule';
import { TasksService } from './tasks.service';

@Module({
  imports: [
    // 仅在实例ID为0的进程中启用定时任务
    process.env.NODE_APP_INSTANCE === '0' ? ScheduleModule.forRoot() : [],
  ],
  providers: [TasksService],
})
export class TasksModule {}

• 或者在定时任务服务中动态控制执行：

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

@Injectable()
export class TasksService implements OnModuleInit {
  private isEnabled = false;

  onModuleInit() {
    // 仅允许实例0执行任务
    this.isEnabled = process.env.NODE_APP_INSTANCE === '0';
  }

  @Cron('0 * * * *')
  async handleCron() {
    if (!this.isEnabled) return;
    // 执行你的任务逻辑
  }
}

方案2：使用分布式锁控制任务执行 #

通过Redis分布式锁保证同一时间只有一个进程能执行定时任务，避免重复操作数据库。

• 引入redlock库，在任务执行前获取锁：

import { Injectable } from '@nestjs/common';
import { Cron } from '@nestjs/schedule';
import Redlock from 'redlock';
import { createClient } from 'redis';

@Injectable()
export class TasksService {
  private redlock: Redlock;

  constructor() {
    const redisClient = createClient({ url: 'redis://localhost:6379' });
    redisClient.connect();
    this.redlock = new Redlock([redisClient], {
      driftFactor: 0.01,
      retryCount: 3,
      retryDelay: 200,
    });
  }

  @Cron('0 * * * *')
  async handleCron() {
    let lock;
    try {
      // 获取锁，有效期10秒（根据任务执行时长调整）
      lock = await this.redlock.lock('task:unique-lock', 10000);
      // 执行你的任务逻辑，此时只有当前进程持有锁
    } catch (err) {
      // 获取锁失败，说明其他进程正在执行任务，直接跳过
      return;
    } finally {
      if (lock) {
        await lock.unlock().catch(() => {});
      }
    }
  }
}

方案3：拆分定时任务为独立单实例进程 #

修改PM2配置，将定时任务服务单独部署为单实例进程，与业务集群分离，从根源避免重复执行。

• 编辑ecosystem.config.js：

module.exports = {
  apps: [
    // 业务集群服务（多实例）
    {
      name: 'business-api',
      script: 'dist/main.js',
      instances: 'max',
      exec_mode: 'cluster',
      env: {
        NODE_ENV: 'production',
        DISABLE_SCHEDULE: 'true',
      },
    },
    // 定时任务服务（单实例）
    {
      name: 'task-scheduler',
      script: 'dist/main.js',
      instances: 1,
      exec_mode: 'fork',
      env: {
        NODE_ENV: 'production',
        ENABLE_SCHEDULE: 'true',
      },
    },
  ],
};

• 在NestJS应用中根据环境变量决定是否启用定时任务：

// main.ts
async function bootstrap() {
  const app = await NestFactory.create(AppModule);
  
  if (process.env.ENABLE_SCHEDULE === 'true') {
    // 仅在定时任务进程中启用调度器
    await app.get(ScheduleRegistry);
  }

  await app.listen(3000);
}
bootstrap();

方案4：数据库层面实现幂等性 #

通过数据库的状态标记、乐观锁或唯一约束，保证即使多个进程触发任务，最终只有一次操作生效。

• 示例：使用状态标记控制任务执行

import { Injectable } from '@nestjs/common';
import { Cron } from '@nestjs/schedule';
import { InjectRepository } from '@nestjs/typeorm';
import { Repository } from 'typeorm';
import { TaskRecord } from './task-record.entity';

@Injectable()
export class TasksService {
  constructor(@InjectRepository(TaskRecord) private repo: Repository<TaskRecord>) {}

  @Cron('0 * * * *')
  async handleCron() {
    // 尝试将一条待执行的任务记录标记为执行中
    const result = await this.repo.update(
      { status: 'pending', executeTime: new Date() },
      { status: 'running' },
      { returning: true }
    );

    // 如果没有更新到记录，说明其他进程已经处理过，直接返回
    if (result.affected === 0) return;

    const task = result.raw[0];
    try {
      // 执行任务逻辑
      // ...
      // 任务完成后更新状态为成功
      await this.repo.update(task.id, { status: 'success' });
    } catch (err) {
      // 任务失败，回滚状态
      await this.repo.update(task.id, { status: 'failed' });
    }
  }
}

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