第 189 章:任务调度(Schedule)与队列

学习目标

• 使用 @nestjs/schedule 实现定时任务
• 集成 BullMQ 队列
• 掌握异步任务处理
• 学会延迟任务与重试

一、定时任务(@nestjs/schedule)

1.1 安装

pnpm add @nestjs/schedule

1.2 注册

import { ScheduleModule } from '@nestjs/schedule';

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

二、@Cron 定时任务

2.1 Cron 表达式

┌──────── 秒 (可选,0-59)
│ ┌────── 分 (0-59)
│ │ ┌──── 时 (0-23)
│ │ │ ┌── 日 (1-31)
│ │ │ │ ┌─ 月 (1-12)
│ │ │ │ │ ┌─ 周 (0-7,0和7都是周日)
│ │ │ │ │ │
* * * * * *

2.2 常用表达式

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

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

  // 每分钟执行
  @Cron('0 * * * * *')
  handleEveryMinute() {
    this.logger.log('每分钟任务');
  }

  // 每 5 分钟
  @Cron('0 */5 * * * *')
  handleEveryFiveMinutes() {}

  // 每天凌晨 3 点
  @Cron('0 0 3 * * *')
  handleDaily() {}

  // 每周一上午 9 点
  @Cron('0 0 9 * * 1')
  handleWeekly() {}

  // 每月 1 号凌晨
  @Cron('0 0 0 1 * *')
  handleMonthly() {}
}

2.3 命名 Cron

@Cron('0 0 3 * * *', { name: 'daily-cleanup' })
dailyCleanup() {}

2.4 时区

@Cron('0 0 9 * * *', {
  name: 'morning-report',
  timeZone: 'Asia/Shanghai',
})
morningReport() {}

三、间隔任务

import { Interval } from '@nestjs/schedule';

@Injectable()
export class TaskService {
  // 每 10 秒
  @Interval('heartbeat', 10_000)
  heartbeat() {
    console.log('heartbeat');
  }
}

四、超时任务

import { Timeout } from '@nestjs/schedule';

@Injectable()
export class TaskService {
  // 启动 5 秒后执行一次
  @Timeout('init', 5_000)
  init() {
    console.log('5 秒后初始化');
  }
}

五、动态任务控制

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

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

  // 动态添加
  addCronJob(name: string, time: string, callback: () => void) {
    const job = new CronJob(time, callback);
    this.schedulerRegistry.addCronJob(name, job);
    job.start();
  }

  // 启动 / 停止
  startCron(name: string) {
    this.schedulerRegistry.getCronJob(name).start();
  }

  stopCron(name: string) {
    this.schedulerRegistry.getCronJob(name).stop();
  }

  // 删除
  deleteCron(name: string) {
    this.schedulerRegistry.deleteCronJob(name);
  }

  // 列出所有
  listCrons() {
    const jobs = this.schedulerRegistry.getCronJobs();
    jobs.forEach((job, name) => {
      console.log(name, job.nextDate().toISOString());
    });
  }
}

六、队列(BullMQ)

6.1 适用场景

• 异步任务(发邮件、推送)
• 延迟任务(订单 30 分钟后关闭)
• 定时任务(更强大,支持重试)
• 任务去重 / 限流

6.2 安装

pnpm add @nestjs/bullmq bullmq

6.3 注册

import { BullModule } from '@nestjs/bullmq';

@Module({
  imports: [
    BullModule.forRoot({
      connection: {
        host: process.env.REDIS_HOST,
        port: Number(process.env.REDIS_PORT),
      },
    }),
    BullModule.registerQueue({
      name: 'email',       // 队列名
    }),
  ],
})
export class AppModule {}

6.4 Producer

import { InjectQueue } from '@nestjs/bullmq';
import { Queue } from 'bullmq';

@Injectable()
export class EmailService {
  constructor(@InjectQueue('email') private emailQueue: Queue) {}

  async sendWelcomeEmail(userId: number) {
    await this.emailQueue.add(
      'welcome',                     // job 名
      { userId },                    // 数据
      {
        attempts: 3,                 // 重试次数
        backoff: { type: 'exponential', delay: 1000 },
        removeOnComplete: true,
        removeOnFail: false,
        delay: 5_000,                // 延迟 5 秒
        priority: 1,                 // 优先级
      },
    );
  }
}

6.5 Consumer

import { Processor, WorkerHost } from '@nestjs/bullmq';
import { Job } from 'bullmq';

@Processor('email')
export class EmailProcessor extends WorkerHost {
  private readonly logger = new Logger(EmailProcessor.name);

  constructor(private mailer: MailerService) {
    super();
  }

  async process(job: Job): Promise<any> {
    this.logger.log(`处理任务: ${job.name}`);

    switch (job.name) {
      case 'welcome':
        await this.sendWelcome(job.data);
        break;
      case 'password-reset':
        await this.sendPasswordReset(job.data);
        break;
    }
  }

  private async sendWelcome(data: { userId: number }) {
    await this.mailer.send({
      to: `user${data.userId}@x.com`,
      subject: '欢迎注册',
      html: '<h1>欢迎</h1>',
    });
  }

  private async sendPasswordReset(data: { email: string }) {
    // ...
  }
}

6.6 监听事件

import { OnQueueActive, OnQueueCompleted, OnQueueFailed } from '@nestjs/bullmq';

@Processor('email')
export class EmailProcessor extends WorkerHost {
  @OnQueueActive()
  onActive(job: Job) {
    console.log(`开始: ${job.id}`);
  }

  @OnQueueCompleted()
  onComplete(job: Job, result: any) {
    console.log(`完成: ${job.id}`, result);
  }

  @OnQueueFailed()
  onFail(job: Job, err: Error) {
    console.error(`失败: ${job.id}`, err);
  }

  async process(job: Job) { /* ... */ }
}

七、延迟队列(订单超时关闭)

@Injectable()
export class OrderService {
  constructor(@InjectQueue('order') private orderQueue: Queue) {}

  async create(dto: CreateOrderDto) {
    const order = await this.orderRepo.save({ ...dto, status: 'pending' });

    // 30 分钟后超时
    await this.orderQueue.add(
      'cancel',
      { orderId: order.id },
      { delay: 30 * 60 * 1000, jobId: `cancel-${order.id}` },  // jobId 避免重复
    );

    return order;
  }

  async pay(orderId: number) {
    await this.orderRepo.update(orderId, { status: 'paid' });

    // 取消超时任务
    const job = await this.orderQueue.getJob(`cancel-${orderId}`);
    if (job) await job.remove();
  }
}

@Processor('order')
export class OrderProcessor extends WorkerHost {
  constructor(private orderRepo: OrderRepository) { super(); }

  async process(job: Job) {
    if (job.name === 'cancel') {
      const order = await this.orderRepo.findOne({ where: { id: job.data.orderId } });
      if (order && order.status === 'pending') {
        await this.orderRepo.update(order.id, { status: 'expired' });
      }
    }
  }
}

八、定时队列(替代 @Cron)

@Injectable()
export class ReportService {
  constructor(@InjectQueue('report') private queue: Queue) {}

  // 每天凌晨 3 点生成报告
  @Cron('0 0 3 * * *')
  scheduleDailyReport() {
    this.queue.add('daily', {}, {
      repeat: { pattern: '0 0 3 * * *' },   // BullMQ 内置 cron
    });
  }
}

九、流量控制

await this.emailQueue.add('welcome', { userId }, {
  // 限流
  limiter: {
    max: 100,           // 100 个
    duration: 60_000,   // 每分钟
  },
});

十、失败重试与死信

@Processor('email', {
  // 失败的 job 重试策略
  defaultJobOptions: {
    attempts: 5,
    backoff: {
      type: 'exponential',
      delay: 2000,
    },
  },
})
export class EmailProcessor extends WorkerHost {
  // 自定义重试逻辑
  async process(job: Job) {
    try {
      await this.sendEmail(job.data);
    } catch (err) {
      if (job.attemptsMade >= 3) {
        // 放死信队列
        await this.handleDeadLetter(job);
      }
      throw err;   // 抛错触发重试
    }
  }
}

十一、@nestjs/schedule vs BullMQ

维度	@nestjs/schedule	BullMQ
简单 cron	✅	✅
集群	❌	✅
重试	❌	✅
延迟任务	❌	✅
UI	❌	bull-board
复杂度	低	中

选择

• 简单定时任务 → @nestjs/schedule
• 异步任务、延迟、重试 → BullMQ

十二、本章小结

API	用途
@Cron	cron 表达式
@Interval	间隔
@Timeout	一次性延时
SchedulerRegistry	动态管理
BullModule	Redis 队列
@Processor	消费者
@InjectQueue	生产者

动手练习

1. 写一个 @Cron 每天凌晨清理过期数据
2. 集成 BullMQ 实现邮件队列
3. 实现订单 30 分钟超时关闭
4. 加失败重试(3 次 + 指数退避)

推荐阅读

• 📖 Schedule
• 📖 BullMQ
• 📖 Cron 表达式

---

下一章:第 190 章:部署与运维 →