TypeORM 的事件系统如何工作?包括实体监听器和订阅者 - 面试题

TypeORM 的事件系统允许开发者在实体操作的生命周期中执行自定义逻辑,提供了强大的扩展能力。

事件类型

1. 实体生命周期事件

TypeORM 提供了以下实体生命周期事件:

BeforeInsert - 在实体插入之前触发
AfterInsert - 在实体插入之后触发
BeforeUpdate - 在实体更新之前触发
AfterUpdate - 在实体更新之后触发
BeforeRemove - 在实体删除之前触发
AfterRemove - 在实体删除之后触发
BeforeSoftRemove - 在实体软删除之前触发
AfterSoftRemove - 在实体软删除之后触发
BeforeRecover - 在实体恢复之前触发
AfterRecover - 在实体恢复之后触发

2. 订阅者事件

订阅者可以监听所有实体的特定事件。

使用实体监听器

基本用法

typescript
import { Entity, PrimaryGeneratedColumn, Column, BeforeInsert, BeforeUpdate, AfterInsert, AfterUpdate } from 'typeorm';

@Entity()
export class User {
  @PrimaryGeneratedColumn()
  id: number;

  @Column()
  name: string;

  @Column()
  email: string;

  @Column({ type: 'timestamp' })
  createdAt: Date;

  @Column({ type: 'timestamp' })
  updatedAt: Date;

  @Column({ default: 0 })
  version: number;

  @BeforeInsert()
  beforeInsert() {
    this.createdAt = new Date();
    this.updatedAt = new Date();
    this.version = 1;
  }

  @BeforeUpdate()
  beforeUpdate() {
    this.updatedAt = new Date();
    this.version++;
  }

  @AfterInsert()
  afterInsert() {
    console.log(`User ${this.name} inserted with ID ${this.id}`);
  }

  @AfterUpdate()
  afterUpdate() {
    console.log(`User ${this.name} updated to version ${this.version}`);
  }
}

复杂逻辑处理

typescript
import { Entity, PrimaryGeneratedColumn, Column, BeforeInsert, BeforeUpdate } from 'typeorm';
import { hash } from 'bcrypt';

@Entity()
export class User {
  @PrimaryGeneratedColumn()
  id: number;

  @Column()
  name: string;

  @Column()
  email: string;

  @Column()
  password: string;

  @Column({ default: false })
  emailVerified: boolean;

  @Column({ type: 'timestamp', nullable: true })
  emailVerifiedAt: Date;

  @Column({ type: 'timestamp' })
  createdAt: Date;

  @Column({ type: 'timestamp' })
  updatedAt: Date;

  @BeforeInsert()
  async beforeInsert() {
    this.createdAt = new Date();
    this.updatedAt = new Date();
    
    // 加密密码
    if (this.password) {
      this.password = await hash(this.password, 10);
    }
    
    // 验证邮箱格式
    if (!this.validateEmail(this.email)) {
      throw new Error('Invalid email format');
    }
  }

  @BeforeUpdate()
  async beforeUpdate() {
    this.updatedAt = new Date();
    
    // 如果密码被修改,重新加密
    if (this.password && this.isPasswordModified()) {
      this.password = await hash(this.password, 10);
    }
    
    // 如果邮箱被验证,记录验证时间
    if (this.emailVerified && !this.emailVerifiedAt) {
      this.emailVerifiedAt = new Date();
    }
  }

  private validateEmail(email: string): boolean {
    const emailRegex = /^[^\s@]+@[^\s@]+\.[^\s@]+$/;
    return emailRegex.test(email);
  }

  private isPasswordModified(): boolean {
    // 实现密码修改检测逻辑
    return true;
  }
}

使用订阅者

基本订阅者

typescript
import { EntitySubscriberInterface, EventSubscriber, InsertEvent, UpdateEvent, RemoveEvent } from 'typeorm';
import { User } from '../entity/User';

@EventSubscriber()
export class UserSubscriber implements EntitySubscriberInterface<User> {
  // 指定监听的实体
  listenTo() {
    return User;
  }

  // 插入前
  beforeInsert(event: InsertEvent<User>) {
    console.log(`Before inserting user: ${event.entity.name}`);
    
    // 可以修改实体
    event.entity.createdAt = new Date();
  }

  // 插入后
  afterInsert(event: InsertEvent<User>) {
    console.log(`After inserting user with ID: ${event.entity.id}`);
    
    // 发送欢迎邮件
    this.sendWelcomeEmail(event.entity);
  }

  // 更新前
  beforeUpdate(event: UpdateEvent<User>) {
    console.log(`Before updating user: ${event.entity.name}`);
    
    // 记录变更
    this.logChanges(event);
  }

  // 更新后
  afterUpdate(event: UpdateEvent<User>) {
    console.log(`After updating user: ${event.entity.name}`);
    
    // 发送通知
    this.sendUpdateNotification(event.entity);
  }

  // 删除前
  beforeRemove(event: RemoveEvent<User>) {
    console.log(`Before removing user: ${event.entity.name}`);
    
    // 检查是否可以删除
    if (event.entity.posts && event.entity.posts.length > 0) {
      throw new Error('Cannot delete user with posts');
    }
  }

  // 删除后
  afterRemove(event: RemoveEvent<User>) {
    console.log(`After removing user: ${event.entity.name}`);
    
    // 清理相关数据
    this.cleanupUserData(event.entity.id);
  }

  private sendWelcomeEmail(user: User) {
    // 发送欢迎邮件逻辑
    console.log(`Sending welcome email to ${user.email}`);
  }

  private sendUpdateNotification(user: User) {
    // 发送更新通知逻辑
    console.log(`Sending update notification to ${user.email}`);
  }

  private logChanges(event: UpdateEvent<User>) {
    // 记录变更逻辑
    console.log('Changes:', event.updatedColumns);
  }

  private cleanupUserData(userId: number) {
    // 清理用户数据逻辑
    console.log(`Cleaning up data for user ${userId}`);
  }
}

全局订阅者

typescript
import { EntitySubscriberInterface, EventSubscriber, InsertEvent } from 'typeorm';

@EventSubscriber()
export class AuditSubscriber implements EntitySubscriberInterface {
  // 监听所有实体
  listenTo() {
    return Object;
  }

  // 所有实体的插入操作
  afterInsert(event: InsertEvent<any>) {
    console.log(`Entity ${event.metadata.name} inserted with ID ${event.entity.id}`);
    
    // 记录审计日志
    this.logAudit({
      action: 'INSERT',
      entity: event.metadata.name,
      entityId: event.entity.id,
      timestamp: new Date(),
    });
  }

  private logAudit(log: any) {
    // 记录审计日志逻辑
    console.log('Audit log:', log);
  }
}

注册订阅者

在 DataSource 中注册

typescript
import { DataSource } from 'typeorm';
import { UserSubscriber } from './subscriber/UserSubscriber';
import { AuditSubscriber } from './subscriber/AuditSubscriber';

const dataSource = new DataSource({
  type: 'mysql',
  host: 'localhost',
  port: 3306,
  username: 'root',
  password: 'password',
  database: 'myapp',
  entities: [User, Post],
  synchronize: false,
  logging: true,
  
  // 注册订阅者
  subscribers: [UserSubscriber, AuditSubscriber],
});

动态注册订阅者

typescript
import { DataSource } from 'typeorm';

const dataSource = new DataSource({
  type: 'mysql',
  host: 'localhost',
  port: 3306,
  username: 'root',
  password: 'password',
  database: 'myapp',
  entities: [User, Post],
  synchronize: false,
  logging: true,
});

// 初始化后动态注册订阅者
dataSource.initialize().then(() => {
  const userSubscriber = new UserSubscriber();
  dataSource.subscribers.push(userSubscriber);
});

高级事件处理

事务中的事件

typescript
@EventSubscriber()
export class TransactionSubscriber implements EntitySubscriberInterface<User> {
  listenTo() {
    return User;
  }

  afterInsert(event: InsertEvent<User>) {
    // 检查是否在事务中
    if (event.queryRunner?.isTransactionActive) {
      console.log('Insert operation is part of a transaction');
    }
    
    // 使用事务执行器
    if (event.queryRunner) {
      event.queryRunner.manager.getRepository(AuditLog).save({
        action: 'USER_INSERT',
        userId: event.entity.id,
        timestamp: new Date(),
      });
    }
  }
}

异步事件处理

typescript
@EventSubscriber()
export class AsyncSubscriber implements EntitySubscriberInterface<User> {
  listenTo() {
    return User;
  }

  async afterInsert(event: InsertEvent<User>) {
    // 异步发送邮件
    await this.sendEmailAsync(event.entity);
    
    // 异步生成用户资料
    await this.generateUserProfileAsync(event.entity);
  }

  private async sendEmailAsync(user: User) {
    // 模拟异步邮件发送
    return new Promise((resolve) => {
      setTimeout(() => {
        console.log(`Email sent to ${user.email}`);
        resolve(null);
      }, 1000);
    });
  }

  private async generateUserProfileAsync(user: User) {
    // 模拟异步用户资料生成
    return new Promise((resolve) => {
      setTimeout(() => {
        console.log(`Profile generated for user ${user.id}`);
        resolve(null);
      }, 500);
    });
  }
}

条件事件处理

typescript
@EventSubscriber()
export class ConditionalSubscriber implements EntitySubscriberInterface<User> {
  listenTo() {
    return User;
  }

  beforeUpdate(event: UpdateEvent<User>) {
    // 只在特定条件下执行
    if (this.shouldProcessUpdate(event)) {
      this.processUpdate(event);
    }
  }

  private shouldProcessUpdate(event: UpdateEvent<User>): boolean {
    // 检查是否更新了特定字段
    const updatedFields = event.updatedColumns.map(col => col.propertyName);
    return updatedFields.includes('email') || updatedFields.includes('password');
  }

  private processUpdate(event: UpdateEvent<User>) {
    // 处理更新逻辑
    console.log('Processing critical update:', event.entity);
  }
}

事件最佳实践

1. 保持事件处理简单

typescript
// ✅ 好的做法:事件处理简单直接
@EventSubscriber()
export class SimpleSubscriber implements EntitySubscriberInterface<User> {
  listenTo() {
    return User;
  }

  afterInsert(event: InsertEvent<User>) {
    // 简单的日志记录
    console.log(`User created: ${event.entity.name}`);
  }
}

// ❌ 不好的做法:事件处理过于复杂
@EventSubscriber()
export class ComplexSubscriber implements EntitySubscriberInterface<User> {
  listenTo() {
    return User;
  }

  async afterInsert(event: InsertEvent<User>) {
    // 复杂的业务逻辑
    const user = event.entity;
    
    // 发送邮件
    await this.sendEmail(user);
    
    // 创建用户资料
    await this.createProfile(user);
    
    // 初始化用户设置
    await this.initializeSettings(user);
    
    // 发送欢迎消息
    await this.sendWelcomeMessage(user);
    
    // 记录统计
    await this.recordStatistics(user);
    
    // 更新缓存
    await this.updateCache(user);
    
    // 触发其他事件
    await this.triggerEvents(user);
  }
}

2. 避免循环事件

typescript
// ✅ 好的做法:避免循环事件
@EventSubscriber()
export class SafeSubscriber implements EntitySubscriberInterface<User> {
  listenTo() {
    return User;
  }

  async afterInsert(event: InsertEvent<User>) {
    // 使用标志位避免循环
    if (event.entity.processed) {
      return;
    }
    
    // 处理逻辑
    await this.processUser(event.entity);
    
    // 标记为已处理
    event.entity.processed = true;
  }
}

// ❌ 不好的做法:可能导致循环事件
@EventSubscriber()
export class CircularSubscriber implements EntitySubscriberInterface<User> {
  listenTo() {
    return User;
  }

  async afterInsert(event: InsertEvent<User>) {
    // 更新用户,可能触发 afterUpdate 事件
    await event.manager.save(User, {
      id: event.entity.id,
      processed: true,
    });
  }
}

3. 错误处理

typescript
// ✅ 好的做法:适当的错误处理
@EventSubscriber()
export class ErrorHandlingSubscriber implements EntitySubscriberInterface<User> {
  listenTo() {
    return User;
  }

  async afterInsert(event: InsertEvent<User>) {
    try {
      await this.sendWelcomeEmail(event.entity);
    } catch (error) {
      console.error('Failed to send welcome email:', error);
      
      // 记录错误,但不影响主流程
      await this.logError(error, event.entity);
    }
  }

  private async logError(error: any, user: User) {
    // 记录错误到数据库
    await event.manager.getRepository(ErrorLog).save({
      error: error.message,
      userId: user.id,
      timestamp: new Date(),
    });
  }
}

4. 性能考虑

typescript
// ✅ 好的做法:批量处理
@EventSubscriber()
export class BatchSubscriber implements EntitySubscriberInterface<User> {
  private batch: User[] = [];
  private timer: NodeJS.Timeout | null = null;

  listenTo() {
    return User;
  }

  afterInsert(event: InsertEvent<User>) {
    // 添加到批次
    this.batch.push(event.entity);
    
    // 设置定时器
    if (!this.timer) {
      this.timer = setTimeout(() => {
        this.processBatch();
      }, 1000); // 1 秒后处理
    }
  }

  private async processBatch() {
    if (this.batch.length === 0) {
      return;
    }
    
    const usersToProcess = [...this.batch];
    this.batch = [];
    this.timer = null;
    
    // 批量处理
    await this.sendBatchNotifications(usersToProcess);
  }

  private async sendBatchNotifications(users: User[]) {
    console.log(`Sending notifications to ${users.length} users`);
    // 批量发送通知逻辑
  }
}

实际应用场景

1. 审计日志

typescript
@EventSubscriber()
export class AuditLogSubscriber implements EntitySubscriberInterface {
  listenTo() {
    return Object;
  }

  afterInsert(event: InsertEvent<any>) {
    this.logAudit('INSERT', event.entity);
  }

  afterUpdate(event: UpdateEvent<any>) {
    this.logAudit('UPDATE', event.entity, event.updatedColumns);
  }

  afterRemove(event: RemoveEvent<any>) {
    this.logAudit('DELETE', event.entity);
  }

  private async logAudit(action: string, entity: any, columns?: any[]) {
    const auditLog = {
      action,
      entityName: entity.constructor.name,
      entityId: entity.id,
      changes: columns ? columns.map(col => col.propertyName) : null,
      timestamp: new Date(),
    };

    await event.manager.getRepository(AuditLog).save(auditLog);
  }
}

2. 缓存失效

typescript
@EventSubscriber()
export class CacheInvalidationSubscriber implements EntitySubscriberInterface<User> {
  listenTo() {
    return User;
  }

  afterUpdate(event: UpdateEvent<User>) {
    // 清除用户缓存
    this.clearUserCache(event.entity.id);
    
    // 清除相关缓存
    this.clearRelatedCache(event.entity.id);
  }

  afterRemove(event: RemoveEvent<User>) {
    // 清除所有相关缓存
    this.clearAllUserCache(event.entity.id);
  }

  private clearUserCache(userId: number) {
    // 清除用户缓存逻辑
    console.log(`Clearing cache for user ${userId}`);
  }

  private clearRelatedCache(userId: number) {
    // 清除相关缓存逻辑
    console.log(`Clearing related cache for user ${userId}`);
  }

  private clearAllUserCache(userId: number) {
    // 清除所有用户缓存逻辑
    console.log(`Clearing all cache for user ${userId}`);
  }
}

3. 通知系统

typescript
@EventSubscriber()
export class NotificationSubscriber implements EntitySubscriberInterface<Post> {
  listenTo() {
    return Post;
  }

  afterInsert(event: InsertEvent<Post>) {
    // 通知关注者
    this.notifyFollowers(event.entity);
    
    // 通知作者
    this.notifyAuthor(event.entity);
  }

  afterUpdate(event: UpdateEvent<Post>) {
    // 如果文章被发布,通知关注者
    if (this.isPublished(event)) {
      this.notifyFollowers(event.entity);
    }
  }

  private isPublished(event: UpdateEvent<Post>): boolean {
    const updatedFields = event.updatedColumns.map(col => col.propertyName);
    return updatedFields.includes('status') && event.entity.status === 'published';
  }

  private async notifyFollowers(post: Post) {
    // 通知关注者逻辑
    console.log(`Notifying followers of post ${post.id}`);
  }

  private async notifyAuthor(post: Post) {
    // 通知作者逻辑
    console.log(`Notifying author of post ${post.id}`);
  }
}

TypeORM 的事件系统提供了强大的扩展能力,合理使用事件可以简化业务逻辑,提高代码的可维护性。