J9 · 事件、定时任务与异步处理

第 9 天 · 11 月 20 日周五 · 交付物：状态变更时发送通知

Spring 提供应用事件、@Scheduled 和 JMS；Nest 用 @nestjs/event-emitter、@nestjs/schedule 和 @nestjs/bullmq 提供同样的积木。区别在于执行模型：一切都跑在 Node 的单线程上，在请求之外抛出的异常不会传给任何人。今天结束时，你会把副作用（通知）从请求中解耦，安排一个夜间任务，并把工作转交给一个带自动重试的 Redis 队列。

目标

用 EventEmitter2 发出业务事件，用 @OnEvent 以同步或异步方式消费它。
用 @Cron、@Interval 和 @Timeout 安排定时任务。
用 @nestjs/bullmq 把处理转交给 Redis 队列：processor、重试次数、退避（backoff）。
知道监听器、定时任务或 job 中抛出的异常去了哪里，并把它记录到日志。
在 TaskFlow 中交付：由 NotificationsListener 消费的 task.status.changed、一个把任务标记为 OVERDUE 的定时任务，以及可选的通知用 BullMQ 队列。

Spring → NestJS

Spring	NestJS	要点
`ApplicationEventPublisher.publishEvent(event)`	`EventEmitter2.emit('task.status.changed', event)`	事件由名字（字符串）标识，而不是由类标识
`@EventListener`	`@OnEvent('task.status.changed')`	监听器的异常不会传回发送方
`@Async` + `@EventListener`	`@OnEvent(名字, { async: true })`	同一个 JS 线程，稍后执行（`setImmediate`）
`@TransactionalEventListener(phase = AFTER_COMMIT)`	把 `emit()` 放在 `await save()` 之后	事件和事务之间没有自动关联
`@EnableScheduling`	`ScheduleModule.forRoot()`	在 `AppModule` 中导入一次
`@Scheduled(cron = "0 0 1 * * *", zone = "Europe/Paris")`	`@Cron('0 01 * * *', { timeZone: 'Europe/Paris' })`	每个实例都会执行，没有 ShedLock 的对应物
`@Scheduled(fixedRate = 60000)`	`@Interval(60_000)`	`setInterval`：不会等上一轮结束
`TaskScheduler.schedule(任务, 时间点)`	`@Timeout(5_000)`	启动后只执行一次
JMS（`@JmsListener`）、Spring Batch	`@nestjs/bullmq`（`@Processor`、`WorkerHost`）	job 持久化在 Redis 中，每个 job 有自己的重试次数和退避

原理

用 @nestjs/event-emitter 实现应用事件

EventEmitterModule.forRoot() 在 AppModule 中导入一次（这个模块默认是全局的），之后就能注入 EventEmitter2，也就是同名库提供的发射器。启动时，在 onApplicationBootstrap 阶段，Nest 遍历 provider 和控制器，为每个带 @OnEvent('名字') 装饰器的方法订阅事件。

emit(名字, payload) 是一个同步调用：EventEmitter2 调用订阅了这个名字的监听器，然后返回 true（至少有一个监听器）或 false （没有）。所以不带选项的监听器会在 emit() 期间开始执行，一直运行到它的第一个 await。加上 { async: true } 后，它会推迟到事件循环的下一轮（setImmediate）：

text

emit() 不带选项       ：监听器（执行到第一个 await）→ emit() 返回 → service 继续执行
emit() + async: true  ：emit() 返回 → service 继续执行 → 监听器（事件循环的下一轮）

靠名字，不靠类型：Spring 根据事件的类选择监听器，这里只看名字。名字拼错不会报任何错，emit() 只是返回 false。把名字放在常量里，把内容放在类里。名字里的点（task.status.changed）只是约定，除非在 forRoot() 里开启 wildcard: true 来监听 task.*。

类比失效的地方：Spring 的同步 @EventListener 在调用方的线程和事务中执行，它的异常会传回 publishEvent()。这里都不成立：事件不关联任何事务，异常会被 Nest 拦截（异步错误）。“异步”不等于“另一个线程”：监听器和其他代码共享 Node 的单线程，在里面做繁重计算会阻塞所有请求（事件循环）。

用 @nestjs/schedule 实现定时任务

ScheduleModule.forRoot() 的作用相当于 @EnableScheduling。三个装饰器用在单例 provider 的方法上：

装饰器	机制	Spring 对应
`@Cron(expression, { name, timeZone })`	`cron` 库	`@Scheduled(cron = ..., zone = ...)`
`@Interval(ms)` 或 `@Interval(名字, ms)`	`setInterval`	`@Scheduled(fixedRate = ...)`
`@Timeout(ms)` 或 `@Timeout(名字, ms)`	`setTimeout`，只执行一次	没有直接对应

任务在应用启动时挂载，在应用关闭时停止。SchedulerRegistry（可注入）按名字找到它们，例如 getCronJob('mark-overdue-tasks')，可以在运行时停止任务或新增任务。

cron 表达式有 5 个字段（第一个是分钟）或 6 个字段（第一个是秒）。CronExpression.EVERY_DAY_AT_1AM 的值是 '0 01 * * *'；Spring 的表达式 0 0 1 * * *（6 个字段）也能用，含义相同。不设 timeZone 时，用的是进程的本地时间，在容器里通常是 UTC。

// TaskFlow 之外的示例：需要在某个模块的 providers 中声明。
import { Injectable, Logger } from '@nestjs/common';
import { Interval, Timeout } from '@nestjs/schedule';

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

  @Timeout(5_000) // 只执行一次，启动后 5 秒
  async warmUp(): Promise<void> {
    this.logger.log('cache warmed up');
  }

  @Interval('refresh-stats', 60_000) // 每 60 秒一次，命名后可在 SchedulerRegistry 中找到
  async refresh(): Promise<void> {
    this.logger.log('stats refreshed');
  }
}

定时方法在事件循环上执行，而不是在线程池里。即使上一轮（异步的）还没结束，@Interval 也会再次调用方法；对 @Cron，选项 waitForCompletion: true 会在上一次还在运行时跳过这次触发。Nest 给每个方法包了一层 try/catch，把错误记录到日志（上下文为 Scheduler）：异常既不会让应用停止，也不会影响后续执行，和 @Scheduled 默认的错误处理器一样。

类比失效的地方：每个实例都会执行自己的定时任务。Spring 也不会协调多个实例（所以才有 ShedLock），Nest 也没有内置的对应方案。有多个副本时：让处理幂等、在数据库加锁、只让一个实例启用定时任务，或者用队列里的计划 job（OVERDUE 定时任务）。

用 @nestjs/bullmq 处理 Redis job

EventEmitter2 的事件只存在于内存中：如果进程重启或邮件发送失败，通知就丢了，也没有任何东西会重放它。BullMQ 队列把每个 job 存进 Redis；由一个 worker 处理，失败时再重试。

组成部分	作用	在 TaskFlow 中
`BullModule.forRootAsync({ useFactory: () => ({ connection }) })`	共享的 Redis 连接	`NotificationsQueueModule`
`BullModule.registerQueue({ name })`	声明一个队列	`notifications` 队列
`@InjectQueue(name)`，然后 `queue.add(job名, data, options)`	生产者	`NotificationsListener`
`extends WorkerHost` 的类上加 `@Processor(name)`，方法 `process(job)`	消费者（worker）	`NotificationsProcessor`

job 的生命周期：如果 process() 抛出异常（或返回的 promise 被拒绝），job 就失败；只要还有剩余重试次数（attempts），BullMQ 会在 backoff 延迟之后重新安排它。指数退避时，延迟为 2^(n-1) × delay：attempts: 5、delay: 1_000 时，重试分别在 1 秒、2 秒、4 秒、8 秒后发起。最后一次尝试之后，job 以 failed 状态留在 Redis 中（没有设置 removeOnFail）。removeOnComplete: 1_000 只保留最近 1 000 个成功的 job。

BullMQ 保证至少一次执行，而不是恰好一次：被中断的 job（处理过程中进程被杀）可能会被重新执行。process() 要写成幂等的。

JMS / Batch ↔ BullMQ：模型就是带重投递的 @JmsListener（生产者、队列、消费者）。Spring Batch（批量读取、步骤、job 续跑）没有直接对应物。类比失效的地方：Redis 和 PostgreSQL 之间没有事务。TaskFlow 在提交之后才添加 job；如果 queue.add() 失败，错误会被记录，通知就丢了。需要强保证时，用 “outbox” 模式：在写数据的同一个事务里写入消息，再由一个进程把它发布出去。

安装：npm i @nestjs/bullmq bullmq ioredis。在 BullMQ 6 中，ioredis 是可选的 peer 依赖：不显式安装的话，应用会在加载队列模块时崩溃（ioredis 陷阱）。

异步错误

异常只会传给等待这个 promise 的一方。没有人在等监听器、定时任务或 job：错误必须就地处理，否则它会丢失，或者让进程退出。

错误发生在哪里	谁在等待？	会发生什么	在 TaskFlow 中
HTTP 处理函数	Nest	全局过滤器，返回 `problem+json` 响应	`AllExceptionsFilter`
`@OnEvent` 监听器	没人	Nest 捕获并记录日志（上下文为 `Event`），调用方毫不知情	`try/catch` + 带任务 id 的日志
`@Cron`、`@Interval`、`@Timeout`	没人	Nest 捕获并记录日志（上下文为 `Scheduler`）	`try/catch` + 明确的信息
BullMQ worker 的 `process()`	BullMQ	job 失败，按 `attempts` 和 `backoff` 重试	让异常抛出去，这正是重试的信号
没有 `await` 的 promise	没人	未处理的拒绝：Node 默认会让进程退出	被 `no-floating-promises` 禁止（忘记 await）

所以监听器抛出的异常不会传回调用方，不管有没有 async: true：即使通知失败，PATCH /v1/tasks/:id/status 也返回 200。这是有意为之（通知不应撤销状态变更），但错误只会出现在日志里。

选项 @OnEvent(名字, { suppressErrors: false }) 会让异常传出来。只有当调用方用 await this.events.emitAsync(...) 等待监听器时（监听器不带 async: true），它才有意义；如果只是普通的 emit()，异常会变成未处理的拒绝，让进程退出。如果调用方需要知道结果，就别用事件：直接调用 service。

在 Spring 中，抛出异常的 @Async 方法最终交给 AsyncUncaughtExceptionHandler，@Scheduled 记录日志后继续：原理相同，没有人在等结果。

TaskFlow 案例

task.status.changed 事件

事件：一个常量名字加一个不可变的类。

// Event names are plain strings: keep them in one constant.
export const TASK_STATUS_CHANGED = 'task.status.changed';

export class TaskStatusChangedEvent {
  constructor(
    readonly taskId: string,
    readonly projectId: string,
    readonly from: TaskStatus,
    readonly to: TaskStatus,
    readonly changedBy: string,
    readonly occurredAt: Date,
  ) {}
}

TasksService 在变更保存后发出它：

async updateStatus(id: string, status: TaskStatus, user: AuthUser): Promise<Task> {
  const task = await this.tasks.findOneBy({ id });
  if (!task) throw new NotFoundException(`Task ${id} not found`);
  await this.projects.assertAccess(task.projectId, user);
  if (task.status === status) return task;

  const from = task.status;
  task.status = status;
  const saved = await this.tasks.save(task);
  // Emit only once the change is saved. Listeners run separately: if one
  // fails, this request still succeeds (see NotificationsListener).
  this.events.emit(
    TASK_STATUS_CHANGED,
    new TaskStatusChangedEvent(task.id, task.projectId, from, status, user.id, this.clock()),
  );
  return saved;
}

emit() 在 await this.tasks.save(task) 之后。save() 在它自己的事务里执行：事件在提交之后才发出，效果和 @TransactionalEventListener(phase = AFTER_COMMIT) 一样。如果在 dataSource.transaction(...) 内部发出，就可能为一条随后被回滚的数据触发通知（事务）。
状态没变就不发事件（提前返回）。
occurredAt 来自注入的时钟 CLOCK（token），测试里可以换成固定日期。
service 不知道谁在监听：加审计或推送都不用改 TasksService。

监听器：

// @EventListener equivalent. { async: true } runs it after the emitting
// request has moved on: an exception here never reaches TasksService.
@Injectable()
export class NotificationsListener {
  private readonly logger = new Logger(NotificationsListener.name);

  constructor(
    // Present only when NotificationsQueueModule is loaded.
    @Optional() @InjectQueue(NOTIFICATIONS_QUEUE) private readonly queue?: Queue<NotificationJob>,
  ) {}

  @OnEvent(TASK_STATUS_CHANGED, { async: true })
  async onStatusChanged(event: TaskStatusChangedEvent): Promise<void> {
    try {
      if (!this.queue) {
        this.logger.log(`[simulated email] task ${event.taskId}: ${event.from} -> ${event.to}`);
        return;
      }
      await this.queue.add(
        'task-status-changed',
        { taskId: event.taskId, from: event.from, to: event.to, recipientId: event.changedBy },
        { attempts: 5, backoff: { type: 'exponential', delay: 1_000 }, removeOnComplete: 1_000 },
      );
    } catch (error) {
      // Nobody awaits this listener, so log here or the error is lost.
      this.logger.error(`Notification for task ${event.taskId} failed`, error instanceof Error ? error.stack : String(error));
    }
  }
}

{ async: true }：方法在事件循环的下一轮执行，即 updateStatus() 交出控制权之后。
没有队列时（没有 REDIS_URL），它记录一封“模拟邮件”。有队列时，它添加一个 job，带 5 次重试和指数退避（BullMQ 队列）。
@Optional() 相当于 @Autowired(required = false)：只有加载了 NotificationsQueueModule，队列才会被注入。
try/catch 把错误连同任务 id 一起记录；没有它，Nest 也会记录错误，但没有这些上下文。

注册它的模块（只有定义了 REDIS_URL 时才加载队列）：

@Module({
  imports: [
    // A function, not the string 'REDIS_URL': the string form only checks
    // that the value is not "false", so an unset variable would still load
    // the queue module and make it try to reach Redis.
    ConditionalModule.registerWhen(NotificationsQueueModule, (env) => Boolean(env.REDIS_URL)),
  ],
  providers: [NotificationsListener],
})
export class NotificationsModule {}

EventEmitterModule.forRoot() 和 ScheduleModule.forRoot() 在 AppModule 中导入。一次状态变更：

http

PATCH /v1/tasks/2b9d…/status
Authorization: Bearer eyJ…
Content-Type: application/json

{ "status": "DONE" }

响应是 200，带上任务（"status": "DONE"），上下文为 NotificationsListener 的日志包含 [simulated email] task 2b9d…: TODO -> DONE。单元测试验证了事件的发出及其内容，不需要启动任何监听器（单元测试）：

it('saves the new status, then emits task.status.changed', async () => {
  repository.findOneBy.mockResolvedValue(Object.assign(new Task(), { id: 't1', projectId: 'p1', status: TaskStatus.Todo }));

  const task = await service.updateStatus('t1', TaskStatus.Done, user);

  expect(task.status).toBe(TaskStatus.Done);
  expect(repository.save).toHaveBeenCalledTimes(1);
  expect(events.emit).toHaveBeenCalledWith(
    TASK_STATUS_CHANGED,
    new TaskStatusChangedEvent('t1', 'p1', TaskStatus.Todo, TaskStatus.Done, 'u1', now),
  );
});

每日 OVERDUE 定时任务

// @Scheduled(cron = "0 0 1 * * *", zone = "Europe/Paris"). Every running
// instance fires it: with several replicas, use a queue or a lock instead.
@Injectable()
export class TasksScheduler {
  private readonly logger = new Logger(TasksScheduler.name);

  constructor(private readonly tasks: TasksService) {}

  @Cron(CronExpression.EVERY_DAY_AT_1AM, { name: 'mark-overdue-tasks', timeZone: APP_TIME_ZONE })
  async markOverdue(): Promise<void> {
    try {
      const count = await this.tasks.markOverdue();
      this.logger.log(`${count} task(s) marked OVERDUE`);
    } catch (error) {
      // Nobody awaits a cron job: catch, or the error is only half visible.
      this.logger.error('Overdue job failed', error instanceof Error ? error.stack : String(error));
    }
  }
}

/** One UPDATE for every late task; returns how many changed. */
async markOverdue(): Promise<number> {
  // The Paris calendar day, like the cron schedule: toISOString() would give
  // the UTC day, one day behind at 01:00 in summer.
  const today = localDate(this.clock());
  const result = await this.tasks.update(
    {
      dueDate: LessThan(today),
      status: In([TaskStatus.Todo, TaskStatus.InProgress]),
      archived: false,
    },
    { status: TaskStatus.Overdue },
  );
  return result.affected ?? 0;
}

每天巴黎时间凌晨 1 点（APP_TIME_ZONE）。名字 mark-overdue-tasks 让你能在 SchedulerRegistry 中找到这个任务。
用一条 UPDATE 处理所有逾期任务（截止日期已过、状态为 TODO 或 IN_PROGRESS、未归档），而不是逐个加载任务。affected 给出被修改的行数。
dueDate 是 date 类型的列：TypeORM 以 'YYYY-MM-DD' 字符串的形式读取和比较它。
这个处理是幂等的：再执行一次也找不到要改的东西，因为已经是 OVERDUE 的任务被排除了。两个副本都执行，也只是多一次查询。不幂等的 job （比如发送汇总邮件）就需要加锁（PostgreSQL 的 pg_try_advisory_lock）、专门的实例，或者 BullMQ 里的计划 job。
批量 UPDATE 不经过 updateStatus()：自动变成 OVERDUE 不会发出 task.status.changed。如果这个变化需要通知，就在 UPDATE 之后发一个事件。
try/catch：Nest 自己也会记录错误，但不会说明是哪个处理失败了。

想在白天就看到定时任务执行，可以临时把表达式换成 CronExpression.EVERY_30_SECONDS（'*/30 * * * * *'）。

定时任务的时区与当天日期

定时任务在巴黎时间凌晨 1 点触发。如果用 toISOString() 计算 today，得到的是 UTC 日期：冬季（UTC+1）巴黎凌晨 1 点对应 UTC 0 点，日期相同；夏季（UTC+2）对应前一天 UTC 23 点，前一天到期的任务要晚一天才会变成 OVERDUE。因此 TaskFlow 在定时任务的时区中计算日期： src/common/time-zone.ts 里的 APP_TIME_ZONE 和 localDate()（带 timeZone 的 Intl.DateTimeFormat）。一个单元测试把时钟固定在 6 月 30 日 UTC 23 点（巴黎 7 月 1 日凌晨 1 点），并验证比较用的日期是 2026-07-01（J10）。

OVERDUE：是状态还是计算结果？

学习计划把 OVERDUE 当作一种状态存储，TaskFlow 照此实现。但这个状态把一个由时间推导出的事实和进度混在了一起：IN_PROGRESS 的任务变成 OVERDUE 后，就丢失了“进行中”这一信息，而且也没有什么能阻止客户端在 PATCH /v1/tasks/:id/status 里发送 OVERDUE。替代方案：只存储进度，在读取时计算“是否逾期”（截止日期已过且状态不是 DONE），不需要定时任务。见领域模型。

通知用 BullMQ 队列（可选）

Redis 是 Compose 中的一个服务，通过 queue profile 启用：

完整的 docker-compose.yml

yaml

# docker compose up -d                 -> PostgreSQL + Adminer for `npm run start:dev`
# docker compose --profile queue up -d -> + Redis for the BullMQ queue (J9)
# docker compose --profile app up -d   -> + the API itself, built from the Dockerfile (J11)
services:
  db:
    image: postgres:16-alpine
    container_name: taskflow-db
    environment:
      POSTGRES_USER: taskflow
      POSTGRES_PASSWORD: taskflow
      POSTGRES_DB: taskflow
    ports:
      - "5432:5432"
    volumes:
      - pgdata:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U taskflow -d taskflow"]
      interval: 5s
      timeout: 3s
      retries: 10

  # Throwaway database for e2e tests: data lives in memory (tmpfs).
  db-test:
    image: postgres:16-alpine
    container_name: taskflow-db-test
    environment:
      POSTGRES_USER: taskflow
      POSTGRES_PASSWORD: taskflow
      POSTGRES_DB: taskflow_test
    ports:
      - "5433:5432"
    tmpfs:
      - /var/lib/postgresql/data

  adminer:
    image: adminer
    ports:
      - "8081:8080"
    depends_on:
      - db

  redis:
    image: redis:7-alpine
    profiles: [queue]
    ports:
      - "6379:6379"

  app:
    build: .
    profiles: [app]
    environment:
      NODE_ENV: production
      PORT: 3000
      DB_HOST: db
      DB_PORT: 5432
      DB_USER: taskflow
      DB_PASSWORD: taskflow
      DB_NAME: taskflow
      # Empty unless exported: the app then refuses to start (validateEnv), while
      # the other services still work without it.
      JWT_SECRET: ${JWT_SECRET:-}
    ports:
      - "3000:3000"
    depends_on:
      db:
        condition: service_healthy

volumes:
  pgdata:

bash

docker compose --profile queue up -d                     # PostgreSQL、Adminer 和 Redis
echo 'REDIS_URL=redis://localhost:6379' >> .env.development
npm run start:dev

队列模块，只有定义了 REDIS_URL 时才加载（见上面的 NotificationsModule）：

// Only loaded when REDIS_URL is set (see NotificationsModule).
@Module({
  imports: [
    BullModule.forRootAsync({
      inject: [ConfigService],
      useFactory: (config: ConfigService) => ({
        connection: { url: config.getOrThrow<string>('REDIS_URL') },
      }),
    }),
    BullModule.registerQueue({ name: NOTIFICATIONS_QUEUE }),
  ],
  providers: [NotificationsProcessor],
  exports: [BullModule],
})
export class NotificationsQueueModule {}

forRootAsync 从 ConfigService 读取 REDIS_URL；变量缺失时，getOrThrow 会在启动时失败（ConfigModule）。
exports: [BullModule] 让 notifications 队列可以在 NotificationsModule 中注入，也就能注入到监听器里。

消费者：

// The consumer side, like a @JmsListener: runs in the background with
// retries and backoff configured on each job (see NotificationsListener).
@Processor(NOTIFICATIONS_QUEUE)
export class NotificationsProcessor extends WorkerHost {
  private readonly logger = new Logger(NotificationsProcessor.name);

  async process(job: Job<NotificationJob>): Promise<void> {
    // A real implementation would call an email provider here; throwing makes
    // BullMQ retry the job.
    this.logger.log(`Email to ${job.data.recipientId}: task ${job.data.taskId} is now ${job.data.to} (attempt ${job.attemptsMade + 1})`);
  }
}

worker 和 API 跑在同一个进程里：繁重的处理同样会阻塞请求。BullMQ 允许以后把它挪到单独的进程中，生产者不用改。整个流程已在 Redis 上端到端验证过：状态变更 → 事件 → 监听器 → 队列 → NotificationsProcessor 的日志（Email to …: task … is now DONE (attempt 1)）。

要观察重试，就让 process() 抛出异常：日志会显示 attempt 1 到 attempt 5，间隔依次为 1、2、4、8 秒，然后 job 变为 failed。要记录最终失败：

// TaskFlow 之外的示例：要加到 NotificationsProcessor 中的方法
// （import { OnWorkerEvent } from '@nestjs/bullmq'）。
@OnWorkerEvent('failed')
onFailed(job: Job<NotificationJob> | undefined, error: Error) {
  this.logger.warn(`Job ${job?.id} failed after ${job?.attemptsMade} attempt(s): ${error.message}`);
}

ConditionalModule：传函数，不要传字符串

ConditionalModule.registerWhen(Module, 'REDIS_URL') 只检查值是不是 "false"：变量不存在时仍会加载模块，模块会尝试连接 Redis。TaskFlow 传的是函数 (env) => Boolean(env.REDIS_URL)。见这个陷阱。

场景速查

变更后要发通知，又不能拖慢请求 → 保存之后 emit()，配合 @OnEvent(名字, { async: true })（事件）。
某个处理要每晚执行 → 带显式 timeZone 和名字的 @Cron（定时任务）。
发送要能扛住重启并自动重试 → 带 attempts 和 backoff 的 BullMQ job（BullMQ 队列）。
三个副本让定时任务执行了三次 → 幂等处理、数据库加锁或队列里的计划 job（schedule）。
监听器的错误在哪儿都看不到 → 在监听器里用 try/catch 并带业务上下文记录日志（异步错误）。
Redis 只在部分环境存在 → 函数形式的 ConditionalModule.registerWhen，消费方用 @Optional()（BullMQ 队列）。
调用方需要知道副作用是否失败 → 不用事件：直接调用 service（异步错误）。
想测试事件是否发出 → mock EventEmitter2，用 toHaveBeenCalledWith（单元测试）。

陷阱

在提交之前发出事件

放在 dataSource.transaction(...) 里的 emit() 会在提交之前发出：监听器可能为一个随后被回滚的修改发通知。在事务结束之后再发事件。

BullMQ 6 没装 ioredis

“BullMQ could not load the optional 'ioredis' package”：显式安装 ioredis。TaskFlow 固定使用 ioredis 5.11.1，因为 TypeORM 1.1 接受的是 5.x 版本（ioredis 陷阱）。

监听器的异常对调用方不可见。 这是默认行为。在监听器里记录日志；如果调用方需要做出反应，就直接调用 service，而不是发事件。
suppressErrors: false 配合 emit()。 异常变成未处理的拒绝，让 Node 退出。这个选项只配合 await emitAsync() 使用。
事件名拼错。 不报错，也没有监听器：emit() 返回 false。使用共享常量（TASK_STATUS_CHANGED）。
在 onModuleInit 中发出事件。 监听器要到 onApplicationBootstrap 时才订阅：事件丢失（生命周期）。
定时任务在每个实例上都执行。 让处理幂等，或者协调各个实例。
@Interval 重叠执行。 如果一轮耗时超过间隔，两次执行会同时进行。对定时任务用 waitForCompletion: true。
时区。 给定时任务设置 timeZone，并在同一时区中计算日期（OVERDUE 定时任务）。
监听器或定时任务里没有 await 的 promise（this.queue.add(...) 没加 await）：try/catch 捕获不到错误。oxlint 的 no-floating-promises 规则会报出来。
在监听器或 processor 里做 CPU 密集的工作。 它会阻塞进程中的所有请求（事件循环）。

练习

[ ] 安装 @nestjs/event-emitter 和 @nestjs/schedule，在 AppModule 中导入 EventEmitterModule.forRoot() 和 ScheduleModule.forRoot()。
[ ] 在 TasksService 中保存之后发出 task.status.changed，并在 NotificationsListener 中消费它，记录一封模拟邮件。
[ ] 让监听器抛出异常：确认 PATCH /v1/tasks/:id/status 仍然返回 200，并在日志中找到这个错误。再去掉 async: true 试一次，比较日志顺序。
[ ] 编写每日定时任务，把逾期任务标记为 OVERDUE，然后用固定时钟测试它（J10），其中包括一个巴黎夏令时凌晨 1 点的时钟。
[ ] 临时把定时任务改成 CronExpression.EVERY_30_SECONDS，观察 task(s) marked OVERDUE 日志。
[ ] 可选：把 Redis 加入 Compose（queue profile），安装 @nestjs/bullmq bullmq ioredis，把通知发送转交给 BullMQ 队列。
[ ] 可选：让 process() 失败，观察 5 次重试和指数退避。
[ ] 去掉 REDIS_URL，确认应用在没有 Redis 时也能启动。
[ ] 在 LEARNINGS.md 里记下与 @EventListener（事务、异常传播）和 @Scheduled（多实例）的不同之处，然后 git tag jour-09。

J9 · 事件、定时任务与异步处理 ​

目标 ​

Spring → NestJS ​

原理 ​

用 @nestjs/event-emitter 实现应用事件 ​

用 @nestjs/schedule 实现定时任务 ​

用 @nestjs/bullmq 处理 Redis job ​

异步错误 ​

TaskFlow 案例 ​

task.status.changed 事件 ​

每日 OVERDUE 定时任务 ​

通知用 BullMQ 队列（可选） ​

场景速查 ​

陷阱 ​

练习 ​

延伸阅读 ​