消息队列与异步任务可靠性

npm i @nestjs/bullmq bullmq ioredis

// src/app.module.ts
import { BullModule } from "@nestjs/bullmq"

@Module({
  imports: [
    BullModule.forRoot({
      connection: { host: "localhost", port: 6379 } // Redis 连接全局复用
    }),
    BullModule.registerQueue({ name: "email" }) // 队列名即契约
  ]
})
export class AppModule {}

// src/email/email.service.ts
import { Inject, Injectable } from "@nestjs/common"
import { InjectQueue } from "@nestjs/bullmq"
import { Queue } from "bullmq"

export interface EmailJob {
  to: string
  subject: string
  body: string
}

@Injectable()
export class EmailService {
  constructor(
    @InjectQueue("email") private readonly emailQueue: Queue<EmailJob>
  ) {}

  async sendWelcome(to: string) {
    await this.emailQueue.add(
      "welcome", // 任务名
      { to, subject: "欢迎注册", body: "..." },
      {
        attempts: 5, // 最多尝试 5 次
        backoff: { type: "exponential", delay: 3000 }, // 间隔 3s、6s、12s、24s、48s
        removeOnComplete: 100, // 完成记录最多保留 100 条，防 Redis 膨胀
        removeOnFail: 1000
      }
    )
  }
}

// src/email/email.processor.ts
import { Logger } from "@nestjs/common"
import { Processor, WorkerHost } from "@nestjs/bullmq"
import { Job } from "bullmq"
import { EmailJob } from "./email.service"

@Processor("email", { concurrency: 5 }) // 同时处理 5 个任务
export class EmailProcessor extends WorkerHost {
  private readonly logger = new Logger(EmailProcessor.name)

  async process(job: Job<EmailJob>): Promise<void> {
    this.logger.log(`任务 ${job.id} 第 ${job.attemptsMade + 1} 次尝试`)
    await mailer.send(job.data) // 抛异常 = 本轮失败，BullMQ 按策略重试
  }
}

// EmailModule 的 providers 中注册 EmailProcessor

// 24 小时后发送回访邮件
await this.emailQueue.add(
  "follow-up",
  { to, subject: "使用得怎么样？" },
  { delay: 24 * 60 * 60 * 1000 }
)

// 队列事件监听：失败告警的最简实现
this.emailQueue.on("failed", (job, err) => {
  this.logger.error(`任务 ${job?.id} 失败：${err.message}`)
})

// 生产者：用业务唯一键做 jobId——BullMQ 对相同 jobId 的 waiting 任务去重
await this.emailQueue.add(
  "welcome",
  { to, subject: "欢迎注册", body: "..." },
  { jobId: `welcome:${userId}` }   // 同一用户重复触发不会重复排队
)

// 消费者：写操作用唯一约束兜底，重复消费变"冲突后放弃"而非"重复生效"
await this.prisma.couponGrant.create({ data: { userId, campaignId } })
// 重复到达时触发唯一约束冲突 → catch 后视为已处理，正常完成

// 生产者
await this.couponQueue.add(
  "birthday-grant",
  { userId },
  { jobId: `birthday:${userId}:${dayjs().format("YYYY-MM-DD")}` } // 键含日期：跨天可再领
)

// 消费者
async process(job: Job<{ userId: string }>): Promise<void> {
  const { userId } = job.data
  try {
    await this.prisma.couponGrant.create({
      data: { userId, campaignId: "birthday", date: new Date() }
    })
  } catch (e) {
    if (e instanceof Prisma.PrismaClientKnownRequestError && e.code === "P2002") {
      return // 唯一约束冲突 = 已发放过：视为成功，避免无谓重试
    }
    throw e  // 其余异常照常抛出，交给 attempts/backoff
  }
}