1. 从“通知”到“解耦”为什么我们需要发布订阅模式最近在重构一个前端项目遇到了一个典型的“面条式代码”问题一个用户登录成功后需要同时更新导航栏的用户头像、刷新侧边栏的未读消息数、跳转页面并且还要向后台发送一个埋点事件。最开始这些逻辑都写在一个巨大的handleLoginSuccess函数里随着功能迭代这个函数越来越臃肿任何一处修改都可能引发意想不到的副作用。这让我下定决心必须引入一种更优雅的通信机制来解耦这些组件和逻辑而发布订阅模式正是解决这类问题的利器。发布订阅模式简单来说就是一种对象间一对多的依赖关系。当一个对象发布者的状态发生改变时所有依赖于它的对象订阅者都会得到通知并自动更新。这就像你关注了一个公众号订阅每当公众号发布新文章发布所有关注者都会收到推送。在 TypeScript 项目中尤其是在构建复杂的前端应用、Node.js 后端服务或任何需要模块间松散耦合的场景时手动实现一个类型安全的发布订阅中心远比直接使用第三方库更能让你理解其精髓并打造出完全贴合自身业务需求的工具。TypeScript 的静态类型系统为我们编写健壮的发布订阅模式提供了绝佳的支持。它能确保我们发布的事件名称、传递的数据结构在编码阶段就受到严格约束从而避免运行时因拼写错误或数据类型不匹配导致的诡异 Bug。接下来我将手把手带你用 TypeScript 从零实现一个功能完备、类型安全的发布订阅器并深入探讨在实际项目中如何应用它以及那些官方文档里不会告诉你的“坑”和技巧。2. 核心设计定义我们的类型安全事件总线在动手写代码之前我们先要明确这个事件总线的核心能力。一个基础的发布订阅器至少需要三个方法on订阅事件、emit发布事件、off取消订阅。进阶功能可能还包括once一次性订阅、清除所有订阅等。但最关键的是我们要用 TypeScript 的泛型来锁定事件名和对应的数据格式。2.1 用泛型约束事件映射表首先我们定义一个事件映射类型EventMap。它将事件名称作为字符串字面量类型与事件触发时传递的数据类型关联起来。这是实现类型安全的核心。// 定义一个泛型接口用于描述所有可能的事件及其对应的数据格式 interface EventMap { // 格式 [事件名]: 事件数据的数据类型 // 例如 userLogin 事件会传递一个包含用户名的对象 [key: string]: any; // 初始使用 any后续会优化 }但上面这个any用得太宽泛了失去了类型约束的意义。更好的做法是让使用这个总线的人自己来定义具体的事件类型。因此我们让EventBus类成为一个泛型类接受一个类型参数T这个T必须扩展自EventMap。// 更严格的 EventMap 基础接口约束键必须是字符串 interface BaseEventMap { [event: string]: unknown; // 使用 unknown 比 any 更安全 } // 泛型事件总线类T 是用户自定义的事件映射类型 class EventBusT extends BaseEventMap Recordstring, unknown { // 内部存储结构一个 Map键是事件名值是该事件对应的回调函数数组 private events: Mapkeyof T, Array(data: T[K]) void new Map(); // 后续方法将在这里实现 }这里有几个关键点T extends BaseEventMap这确保了T必须是一个键为字符串、值为任意类型的对象。 Recordstring, unknown这是默认类型参数如果用户不提供T则使用一个所有字符串键对应unknown类型的空映射作为安全兜底。private events: Mapkeyof T, Array(data: T[K]) void这是核心存储结构。keyof T表示T的所有键的联合类型即所有有效事件名。对于每个事件名我们存储一个回调函数数组。回调函数的参数类型T[K]是 TypeScript 的索引访问类型表示事件名K在T映射中对应的值类型。这保证了订阅userLogin事件的回调其参数类型一定是T[userLogin]。2.2 实现核心的订阅on方法订阅方法需要接收一个事件名和一个回调函数。回调函数的参数类型必须与事件名所映射的数据类型匹配。class EventBusT extends BaseEventMap Recordstring, unknown { private events: Mapkeyof T, Array(data: T[K]) void new Map(); /** * 订阅事件 * param event 事件名称必须是已定义事件映射 T 中的键 * param callback 事件触发时的回调函数接收对应事件的数据 */ onK extends keyof T(event: K, callback: (data: T[K]) void): void { // 如果该事件还没有订阅者数组则初始化一个空数组 if (!this.events.has(event)) { this.events.set(event, []); } // 将回调函数推入对应事件的数组 this.events.get(event)!.push(callback); } }这里使用了泛型约束K extends keyof T它告诉 TypeScriptevent参数必须是T的某个有效键。同时callback的参数类型被精确地推断为T[K]。当你调用bus.on(userLogin, (data) ...)时data的类型会自动被推断为T[userLogin]编辑器会提供完美的代码补全和类型检查。2.3 实现安全的发布emit方法发布方法需要接收一个事件名和对应的数据。数据必须符合该事件定义的类型。/** * 发布事件 * param event 要触发的事件名称 * param data 传递给所有订阅者的数据类型必须匹配 T[K] */ emitK extends keyof T(event: K, data: T[K]): void { // 获取该事件的所有订阅回调 const callbacks this.events.get(event); // 如果有订阅者则逐一同步调用回调函数 if (callbacks) { // 使用 slice() 创建副本后遍历防止在回调中取消订阅导致数组迭代错乱 callbacks.slice().forEach(callback { try { callback(data); } catch (error) { // 非常重要避免单个回调的错误导致整个事件链崩溃 console.error(Error in event handler for ${String(event)}:, error); } }); } }这里有一个非常重要的实践细节callbacks.slice().forEach(...)。我们为什么需要slice()想象一下如果在某个回调函数内部调用了off方法来取消自身的订阅它会直接修改callbacks数组。在 JavaScript 的forEach迭代过程中修改正在遍历的数组会导致不可预期的行为比如跳过某些回调。通过slice()创建一份浅拷贝我们就在一个稳定的快照上执行迭代确保了所有订阅者都能被正确通知到。另外用try...catch包裹每个回调的执行也是生产环境必备的防御性编程。不能让一个订阅者的错误“炸掉”整个事件流影响其他无关的订阅者。2.4 实现取消订阅off方法取消订阅需要能够移除特定的回调函数。这里我们面临一个挑战如何比较两个函数是否相等在 JavaScript 中函数是引用类型直接比较callback someFunction是可行的但前提是取消订阅时传入的是同一个函数引用。这也是为什么我们通常建议将回调函数定义为变量或类方法而不是直接在on方法里写匿名函数。/** * 取消事件订阅 * param event 事件名称 * param callback 要移除的具体回调函数引用。如果不传则移除该事件的所有订阅。 */ offK extends keyof T(event: K, callback?: (data: T[K]) void): void { const callbacks this.events.get(event); if (!callbacks) { return; // 该事件本来就没有订阅者直接返回 } if (callback) { // 移除特定的回调函数 const index callbacks.indexOf(callback); if (index -1) { callbacks.splice(index, 1); } // 如果该事件的所有回调都被移除了则从 Map 中删除这个键避免内存泄漏 if (callbacks.length 0) { this.events.delete(event); } } else { // 未指定具体回调则清空该事件的所有订阅 this.events.delete(event); } }这里有一个容易忽略的内存泄漏点当一个事件的所有回调都被移除后callbacks数组变成了空数组[]但这个空数组仍然被Map引用着。虽然它不占多少内存但在一个长期运行的单页应用SPA中成千上万个这样无用的空数组和Map键会逐渐累积。因此我们在callbacks.length 0时主动调用this.events.delete(event)来清理资源这是一个良好的编程习惯。3. 进阶功能与边界情况处理一个基础的三件套on,emit,off已经能解决80%的问题。但要让我们的EventBus更健壮、更易用还需要考虑一些进阶场景和边界情况。3.1 实现一次性订阅onceonce方法允许一个回调函数只被调用一次触发后自动取消订阅。这个功能在监听初始化完成、弹窗确认等场景非常有用。实现它的关键在于我们需要包装用户传入的回调函数。/** * 订阅事件仅触发一次 * param event 事件名称 * param callback 一次性回调函数 */ onceK extends keyof T(event: K, callback: (data: T[K]) void): void { // 定义一个包装函数 const wrapper (data: T[K]) { // 先执行用户的回调 callback(data); // 执行完毕后立即取消订阅这个包装函数自身 this.off(event, wrapper); }; // 订阅包装函数 this.on(event, wrapper); }这里有一个精妙之处我们订阅的是内部定义的wrapper函数而不是用户传入的callback。当事件触发时wrapper被调用它先执行用户的逻辑然后调用this.off(event, wrapper)将自己从订阅列表中移除。这个实现清晰且高效。3.2 处理异步回调与执行顺序我们的emit方法是同步执行的所有回调会按照订阅顺序依次、立即执行。这在大多数情况下是符合预期的。但有时订阅者可能包含异步操作比如发起一个网络请求。如果后续的回调依赖于前一个异步操作的结果同步执行就会出问题。对于这种情况EventBus本身通常不负责处理异步流程它只是一个通知机制。更合理的架构是让异步逻辑在回调内部处理或者使用更高级的模式如响应式编程库 RxJS。不过我们可以提供一个简单的异步发射器作为可选方案/** * 异步发布事件等待所有回调可能是异步的完成 * param event 事件名称 * param data 事件数据 * returns Promise在所有回调执行完毕后 resolve */ async emitAsyncK extends keyof T(event: K, data: T[K]): Promisevoid { const callbacks this.events.get(event); if (!callbacks) { return; } // 使用 Promise.all 等待所有回调执行完毕 // 注意这里假设回调可能返回 Promise。如果回调是同步函数也会被包装成 resolved Promise。 const promises callbacks.slice().map(callback { try { const result callback(data); // 如果回调返回了 Promise则等待它否则直接视为完成 return Promise.resolve(result); } catch (error) { // 同步错误仍然返回一个 rejected Promise console.error(Error in async event handler for ${String(event)}:, error); return Promise.reject(error); } }); await Promise.all(promises); }使用emitAsync时需要注意它会等待所有订阅者的回调完成。如果某个回调卡住了比如一个无限循环整个emitAsync调用也会被挂起。因此它更适合用于需要明确知道“所有处理都已完成”的场景比如在关闭应用前保存所有数据。3.3 防御非法事件名与数据尽管 TypeScript 在编译时能帮我们检查类型但运行时仍然可能通过动态代码或第三方库传入非法值。增加一些运行时检查可以增强鲁棒性。onK extends keyof T(event: K, callback: (data: T[K]) void): void { // 基础类型检查 if (typeof event ! string typeof event ! symbol) { throw new TypeError(Event name must be a string or symbol, got ${typeof event}); } if (typeof callback ! function) { throw new TypeError(Callback must be a function); } // ... 原有逻辑 } emitK extends keyof T(event: K, data: T[K]): void { if (typeof event ! string typeof event ! symbol) { // 可以选择静默失败或抛出错误取决于你的设计哲学 console.warn(Attempted to emit event with invalid name: ${event}); return; } // ... 原有逻辑 }对于data参数在 TypeScript 中我们很难做精细的运行时类型验证除非引入像zod这样的验证库。一个折中的办法是在关键的、数据格式复杂的业务事件发布前在调用emit的地方进行数据校验。4. 实战应用在前端框架与Node.js中的集成理论讲完了我们来点实际的。这个EventBus类如何融入不同的技术栈4.1 在 Vue 3 组件中作为响应式事件中心Vue 3 的 Composition API 和provide/inject机制与我们的EventBus是天作之合。我们可以创建一个全局的事件总线并在任何组件中注入使用。首先定义我们应用的事件类型// types/events.ts export interface AppEvents { user:login: { username: string; userId: number }; user:logout: undefined; // 有些事件不需要数据 notification:show: { message: string; type: success | error | info }; cart:updated: { itemCount: number }; // ... 更多事件 }然后创建全局总线实例并注入// utils/eventBus.ts import { EventBus } from ./EventBus; // 我们之前写的类 import type { AppEvents } from ../types/events; // 创建全局单例 export const globalEventBus new EventBusAppEvents(); // 在 Vue 应用的根组件中 provide // app.vue 或 main.ts import { createApp } from vue; import { provide } from vue; import { globalEventBus } from ./utils/eventBus; const app createApp(App); // 提供一个唯一的 InjectionKey const EventBusKey Symbol(event-bus); app.provide(EventBusKey, globalEventBus); app.mount(#app);在任何子组件中使用!-- SomeChildComponent.vue -- script setup langts import { inject, onMounted, onUnmounted } from vue; import type { AppEvents } from ../types/events; import { EventBus } from ../utils/EventBus; import { EventBusKey } from ../symbols; // 需要导出上面定义的 Symbol // 注入事件总线 const bus injectEventBusAppEvents(EventBusKey)!; onMounted(() { // 订阅事件享受完整的类型提示 bus.on(user:login, (userData) { console.log(欢迎回来${userData.username}); // userData.userId 是 number 类型 }); bus.on(notification:show, (notification) { // notification.type 只能是 success | error | info 之一 showToast(notification.message, notification.type); }); }); onUnmounted(() { // 组件卸载时取消所有订阅以避免内存泄漏 // 注意这里需要引用具体的回调函数匿名函数无法取消。 // 更好的做法是将回调定义为组件方法。 }); const handleButtonClick () { // 发布事件 bus.emit(cart:updated, { itemCount: 5 }); // 类型安全 }; /script这种模式完美地将组件解耦。一个位于组件树顶层的导航栏组件可以订阅user:login来更新用户信息而一个深埋在模态框里的结算组件可以发布cart:updated事件两者无需知道彼此的存在。4.2 在 Node.js 后端服务中管理模块间通信在 Node.js 后端发布订阅模式同样大有用武之地。例如在一个电商系统中当订单创建成功后可能需要触发一系列后续动作更新库存、发送邮件通知、记录审计日志、推送消息到消息队列。如果没有事件总线你可能会在订单服务里直接调用库存服务、邮件服务等形成紧密耦合。用上我们的EventBus代码会清晰很多// services/orderService.ts import { eventBus } from ../core/eventBus; // 全局总线实例 import type { SystemEvents } from ../types/systemEvents; interface OrderData { orderId: string; userId: number; items: Array{ productId: string; quantity: number }; } export class OrderService { async createOrder(orderData: OrderData): Promisevoid { // 1. 核心业务逻辑创建订单记录 const order await this.orderRepository.save(orderData); // 2. 发布“订单创建”事件不关心谁监听、如何处理 eventBus.emit(order:created, { orderId: order.id, userId: order.userId, items: order.items, createdAt: new Date(), }); // 3. 返回结果主流程结束 } } // services/inventoryService.ts import { eventBus } from ../core/eventBus; export class InventoryService { constructor() { // 订阅订单创建事件负责扣减库存 eventBus.on(order:created, async (order) { for (const item of order.items) { await this.decreaseStock(item.productId, item.quantity); } console.log(库存已为订单 ${order.orderId} 更新); }); } private async decreaseStock(productId: string, quantity: number): Promisevoid { // 扣减库存逻辑 } } // services/notificationService.ts import { eventBus } from ../core/eventBus; export class NotificationService { constructor() { // 同样订阅订单创建事件负责发邮件 eventBus.on(order:created, async (order) { await this.sendOrderConfirmationEmail(order.userId, order.orderId); }); } }这样OrderService的职责变得非常单一和清晰创建订单并发出事件。库存、邮件等后续处理由各自的服务监听事件后异步完成。如果需要新增一个“给用户增加积分”的功能只需要新建一个RewardService并订阅order:created事件即可完全不需要修改OrderService的代码。这就是发布订阅模式带来的“开闭原则”好处。5. 性能考量、内存管理与调试技巧任何工具引入都需要权衡利弊。发布订阅模式虽然解耦了代码但也引入了一些新的复杂性和潜在风险。5.1 潜在的内存泄漏与订阅管理这是使用事件总线最容易踩的坑。如果一个组件或对象订阅了事件但在销毁时没有取消订阅那么它以及它可能引用的外部作用域闭包将无法被垃圾回收因为事件总线还持有对它的回调函数的引用。最佳实践始终配对on和off尤其是在前端框架React, Vue, Angular中必须在组件的生命周期结束点如onUnmounted,componentWillUnmount,ngOnDestroy取消订阅。// Vue 3 Composition API 示例 import { onUnmounted } from vue; const handleUserLogin (data) { /* ... */ }; bus.on(user:login, handleUserLogin); onUnmounted(() { bus.off(user:login, handleUserLogin); // 必须传入同一个函数引用 });陷阱匿名函数无法被取消订阅// 错误示例匿名函数导致无法取消订阅 bus.on(user:login, (data) { console.log(data); }); // 稍后... 这行代码无效因为传入的不是同一个函数引用 bus.off(user:login, (data) { console.log(data); }); // 正确做法使用具名函数或类方法 const loginHandler (data) { console.log(data); }; bus.on(user:login, loginHandler); // ... bus.off(user:login, loginHandler); // 成功取消为了辅助调试可以给我们的EventBus增加一个简单的调试方法用于查看当前所有活跃的订阅class EventBusT extends BaseEventMap { // ... 其他代码 /** * 获取当前所有事件的订阅者数量调试用 */ getSubscriptionStats(): Recordstring, number { const stats: Recordstring, number {}; for (const [event, callbacks] of this.events.entries()) { stats[String(event)] callbacks.length; } return stats; } /** * 清除所有订阅谨慎使用通常在测试或应用重置时调用 */ clearAll(): void { this.events.clear(); } }5.2 循环依赖与无限循环发布订阅模式可能隐藏循环依赖。例如事件 A 的处理函数中发布了事件 B而事件 B 的处理函数中又发布了事件 A。如果没有适当的防护这会导致无限循环和栈溢出。防护策略避免在事件处理函数中发布可能触发自身的事件。这需要良好的架构设计和对事件流的清晰理解。使用“已处理”标志或防抖。对于可能循环的场景可以在处理函数入口设置一个标志位如果发现当前正在处理该事件则直接返回。将emit改为异步。使用setTimeout或queueMicrotask将事件的发布推到下一个事件循环可以打破同步的循环链但并不能从根本上解决问题只是将栈溢出变成了无限异步循环。// 一个简单的防循环示例不适用于所有场景 class SafeEventBusT extends BaseEventMap extends EventBusT { private emittingEvents new Setkeyof T(); emitK extends keyof T(event: K, data: T[K]): void { // 如果当前正在发布这个事件则跳过防止直接循环 if (this.emittingEvents.has(event)) { console.warn(Potential circular emission detected for event: ${String(event)}); return; } this.emittingEvents.add(event); try { super.emit(event, data); } finally { this.emittingEvents.delete(event); } } }5.3 类型安全的极限与变通方案我们的泛型实现要求所有事件类型必须在EventMap接口中预先定义。这在大型项目中可能有些繁琐而且如果你需要动态地、在运行时注册新的事件类型静态类型系统就无能为力了。对于需要高度动态性的场景可以考虑以下变通方案方案一使用更宽松但仍有约束的类型// 允许任意字符串作为事件名但数据必须是某种已知类型 interface DynamicEventMap { // 所有事件的数据都必须是这个联合类型之一 [key: string]: string | number | boolean | object | undefined; } const bus new EventBusDynamicEventMap(); bus.emit(anyStringEvent, { custom: data }); // 允许 // 缺点是回调函数中的 data 类型不够精确。方案二混合模式保留核心事件的类型安全在实际项目中更常见的做法是区分“核心业务事件”和“通用/调试事件”。核心事件使用严格的类型定义通用事件则使用宽松的类型或any。interface CoreEvents { user:login: UserData; order:created: OrderData; } // 使用交叉类型让总线同时支持严格类型和宽松类型 class HybridEventBusT extends BaseEventMap extends EventBusT Recordstring, any { // 可以添加一些处理动态事件的特殊方法 emitDynamic(event: string, data: any): void { this.emit(event as any, data); // 使用类型断言绕过检查 } } const bus new HybridEventBusCoreEvents(); bus.emit(user:login, { username: test }); // 类型安全 bus.emitDynamic(custom:debug:log, { message: something }); // 动态事件6. 测试策略如何确保事件总线可靠运行为EventBus编写单元测试至关重要它能确保核心逻辑的稳定并在重构时给予我们信心。6.1 基础功能测试使用 Jest 或 Vitest 等测试框架我们可以轻松测试订阅、发布和取消订阅。// EventBus.test.ts import { EventBus } from ./EventBus; interface TestEvents { test:event: { value: number }; another:event: string; } describe(EventBus, () { let bus: EventBusTestEvents; beforeEach(() { bus new EventBusTestEvents(); }); test(should call subscriber when event is emitted, () { const mockCallback jest.fn(); bus.on(test:event, mockCallback); const testData { value: 42 }; bus.emit(test:event, testData); expect(mockCallback).toHaveBeenCalledTimes(1); expect(mockCallback).toHaveBeenCalledWith(testData); }); test(should call multiple subscribers, () { const mockCallback1 jest.fn(); const mockCallback2 jest.fn(); bus.on(test:event, mockCallback1); bus.on(test:event, mockCallback2); bus.emit(test:event, { value: 1 }); expect(mockCallback1).toHaveBeenCalledTimes(1); expect(mockCallback2).toHaveBeenCalledTimes(1); }); test(should not call subscriber after off, () { const mockCallback jest.fn(); bus.on(test:event, mockCallback); bus.off(test:event, mockCallback); bus.emit(test:event, { value: 1 }); expect(mockCallback).not.toHaveBeenCalled(); }); test(once should call callback only once, () { const mockCallback jest.fn(); bus.once(test:event, mockCallback); bus.emit(test:event, { value: 1 }); bus.emit(test:event, { value: 2 }); // 第二次发射回调不应被调用 expect(mockCallback).toHaveBeenCalledTimes(1); expect(mockCallback).toHaveBeenCalledWith({ value: 1 }); }); });6.2 错误处理与边界测试测试异常情况下的行为确保系统的健壮性。test(should not break on subscriber error, () { const consoleSpy jest.spyOn(console, error).mockImplementation(); const goodCallback jest.fn(); const badCallback () { throw new Error(Callback error!); }; bus.on(test:event, badCallback); bus.on(test:event, goodCallback); // 即使第一个回调出错第二个回调也应该被执行 expect(() bus.emit(test:event, { value: 1 })).not.toThrow(); expect(goodCallback).toHaveBeenCalledTimes(1); expect(consoleSpy).toHaveBeenCalled(); // 确认错误被捕获并打印 consoleSpy.mockRestore(); }); test(should handle off for non-existent event or callback gracefully, () { const callback jest.fn(); // 取消一个不存在的事件的订阅不应报错 expect(() bus.off(non:existent as any, callback)).not.toThrow(); // 取消一个从未订阅过的回调不应报错 bus.on(test:event, callback); expect(() bus.off(test:event, () {})).not.toThrow(); // 传入不同的函数引用 });6.3 集成测试示例在更接近真实场景的集成测试中我们可以模拟两个服务通过事件总线通信。// OrderService.integration.test.ts import { EventBus } from ./EventBus; import { OrderService } from ./OrderService; import { InventoryService } from ./InventoryService; interface IntegrationEvents { order:created: { orderId: string; productId: string; quantity: number }; } describe(Order and Inventory Integration, () { let bus: EventBusIntegrationEvents; let orderService: OrderService; let inventoryService: InventoryService; let mockDecreaseStock: jest.Mock; beforeEach(() { bus new EventBusIntegrationEvents(); // 假设这些服务接收一个事件总线实例作为依赖注入 orderService new OrderService(bus); inventoryService new InventoryService(bus); mockDecreaseStock jest.fn(); // 假设我们有办法模拟库存服务的方法 inventoryService.decreaseStock mockDecreaseStock; }); test(creating an order should trigger inventory update, async () { await orderService.createOrder({ orderId: order-123, productId: prod-456, quantity: 2, }); // 验证库存服务的扣减方法被以正确的参数调用 expect(mockDecreaseStock).toHaveBeenCalledTimes(1); expect(mockDecreaseStock).toHaveBeenCalledWith(prod-456, 2); // 注意这里测试的是行为调用了方法而不是直接监听事件。 // 事件总线是通信机制我们关心的是最终效果。 }); });通过这样从单元到集成的测试覆盖你可以自信地重构EventBus的内部实现只要公开的 API 和类型定义不变所有依赖它的代码都能继续正常工作。