服务端阅读 02月18日 22:20
TypeORM 的事件系统如何工作?包括实体监听器和订阅者
TypeORM 的事件系统允许开发者在实体操作的生命周期中执行自定义逻辑,提供了强大的扩展能力。事件类型1. 实体生命周期事件TypeORM 提供了以下实体生命周期事件:BeforeInsert - 在实体插入之前触发AfterInsert - 在实体插入之后触发BeforeUpdate - 在实体更新之前触发AfterUpdate - 在实体更新之后触发BeforeRemove - 在实体删除之前触发AfterRemove - 在实体删除之后触发BeforeSoftRemove - 在实体软删除之前触发AfterSoftRemove - 在实体软删除之后触发BeforeRecover - 在实体恢复之前触发AfterRecover - 在实体恢复之后触发2. 订阅者事件订阅者可以监听所有实体的特定事件。使用实体监听器基本用法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}`); }}复杂逻辑处理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; }}使用订阅者基本订阅者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}`); }}全局订阅者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 中注册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],});动态注册订阅者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);});高级事件处理事务中的事件@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(), }); } }}异步事件处理@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); }); }}条件事件处理@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. 保持事件处理简单// ✅ 好的做法:事件处理简单直接@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. 避免循环事件// ✅ 好的做法:避免循环事件@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. 错误处理// ✅ 好的做法:适当的错误处理@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. 性能考虑// ✅ 好的做法:批量处理@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. 审计日志@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. 缓存失效@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. 通知系统@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 的事件系统提供了强大的扩展能力,合理使用事件可以简化业务逻辑,提高代码的可维护性。