事件流控制与观察者模式

Site Owner

发布于 2026-05-21

系统讲解DOM事件流捕获冒泡机制、事件委托红利、非冒泡事件处理策略、passive监听器及EventEmitter实现

事件流控制与观察者模式

事件驱动编程是 JavaScript 的基因。从 DOM 事件冒泡/捕获、Node.js EventEmitter，到 RxJS 的响应式流，事件管道模式贯穿前后端开发。本文深入实现观察者模式、发布订阅模式、事件委托、以及类 RxJS 的操作符管道。

1. DOM 事件模型三阶段

// 代码示例 1：捕获 → 目标 → 冒泡
document.getElementById('parent').addEventListener('click', () => {
  console.log('parent capture');
}, true); // 捕获阶段

.().(,  {
  .();
}); 

.().(,  {
  .();
}); 




element.(,  {
  e.();          
  e.(); 
});

JavaScript事件流与观察者模式—从DOM事件委托到发布订阅系统|新宇宙博客

// 代码示例 2：生产级 EventEmitter
class EventEmitter {
  #events = new Map();
  #maxListeners = 10;
  
  on(event, listener, options = {}) {
    if (!this.#events.has(event)) this.#events.set(event, []);
    const listeners = this.#events.get(event);
    
    if (listeners.length >= this.#maxListeners) {
      console.warn(`MaxListenersExceededWarning: ${event}`);
    }
    
    const entry = { fn: listener, once: options.once || false, priority: options.priority || 0 };
    listeners.push(entry);
    listeners.sort((a, b) => b.priority - a.priority);
    
    return this;
  }
  
  once(event, listener) {
    return this.on(event, listener, { once: true });
  }
  
  off(event, listener) {
    const listeners = this.#events.get(event);
    if (!listeners) return this;
    const idx = listeners.findIndex(e => e.fn === listener);
    if (idx !== -1) listeners.splice(idx, 1);
    return this;
  }
  
  emit(event, ...args) {
    const listeners = this.#events.get(event);
    if (!listeners || listeners.length === 0) return false;
    
    const toRemove = [];
    for (const entry of [...listeners]) {
      entry.fn(...args);
      if (entry.once) toRemove.push(entry);
    }
    
    for (const entry of toRemove) {
      const idx = listeners.indexOf(entry);
      if (idx !== -1) listeners.splice(idx, 1);
    }
    
    return true;
  }
  
  removeAllListeners(event) {
    if (event) this.#events.delete(event);
    else this.#events.clear();
    return this;
  }
  
  listenerCount(event) {
    return this.#events.get(event)?.length || 0;
  }
  
  // 异步版本
  async emitAsync(event, ...args) {
    const listeners = this.#events.get(event) || [];
    for (const entry of [...listeners]) {
      await entry.fn(...args);
      if (entry.once) this.off(event, entry.fn);
    }
  }
  
  // 等待事件
  waitFor(event, timeout = 0) {
    return new Promise((resolve, reject) => {
      const timer = timeout ? setTimeout(() => {
        this.off(event, handler);
        reject(new Error(`Timeout waiting for ${event}`));
      }, timeout) : null;
      
      const handler = (...args) => {
        if (timer) clearTimeout(timer);
        resolve(args.length === 1 ? args[0] : args);
      };
      
      this.once(event, handler);
    });
  }
}

// 代码示例 3：发布订阅模式（解耦版）
class PubSub {
  static #channels = new Map();
  
  static subscribe(channel, handler) {
    if (!this.#channels.has(channel)) this.#channels.set(channel, new Set());
    this.#channels.get(channel).add(handler);
    return () => this.#channels.get(channel).delete(handler); // unsubscribe
  }
  
  static publish(channel, data) {
    const handlers = this.#channels.get(channel);
    if (handlers) handlers.forEach(h => h(data));
  }
}

// 观察者模式（直接依赖）
class Subject {
  #observers = new Set();
  
  attach(observer) { this.#observers.add(observer); }
  detach(observer) { this.#observers.delete(observer); }
  notify(data) { this.#observers.forEach(o => o.update(data)); }
}

// 代码示例 4：高性能事件委托
function delegate(parent, selector, event, handler) {
  parent.addEventListener(event, function(e) {
    const target = e.target.closest(selector);
    if (target && parent.contains(target)) {
      handler.call(target, e);
    }
  });
}

// 使用：1000 行列表只需 1 个监听器
delegate(document.getElementById('list'), '.list-item', 'click', function(e) {
  console.log('Clicked:', this.dataset.id);
});

// 代码示例 5：AbortController 管理多个监听器
function setupListeners(element) {
  const controller = new AbortController();
  const { signal } = controller;
  
  element.addEventListener('click', handleClick, { signal });
  element.addEventListener('mouseover', handleHover, { signal });
  element.addEventListener('keydown', handleKey, { signal });
  window.addEventListener('resize', handleResize, { signal });
  
  // 一次性清除所有监听器
  return () => controller.abort();
}

// 代码示例 6：简化 Observable 实现
class Observable {
  #subscribe;
  
  constructor(subscribe) {
    this.#subscribe = subscribe;
  }
  
  subscribe(observer) {
    const normalizedObserver = typeof observer === 'function'
      ? { next: observer, error: e => console.error(e), complete: () => {} }
      : { next: () => {}, error: e => console.error(e), complete: () => {}, ...observer };
    
    const subscription = this.#subscribe(normalizedObserver);
    return { unsubscribe: subscription || (() => {}) };
  }
  
  // 创建工厂
  static fromEvent(element, event) {
    return new Observable(observer => {
      const handler = e => observer.next(e);
      element.addEventListener(event, handler);
      return () => element.removeEventListener(event, handler);
    });
  }
  
  static interval(ms) {
    return new Observable(observer => {
      let count = 0;
      const id = setInterval(() => observer.next(count++), ms);
      return () => clearInterval(id);
    });
  }
  
  static from(iterable) {
    return new Observable(observer => {
      for (const item of iterable) {
        observer.next(item);
      }
      observer.complete();
    });
  }
}

// 代码示例 7：链式操作符
Observable.prototype.pipe = function(...operators) {
  return operators.reduce((source, op) => op(source), this);
};

// 操作符工厂
const map = (fn) => (source) => new Observable(observer => {
  return source.subscribe({
    next: value => observer.next(fn(value)),
    error: err => observer.error(err),
    complete: () => observer.complete()
  }).unsubscribe;
});

const filter = (pred) => (source) => new Observable(observer => {
  return source.subscribe({
    next: value => { if (pred(value)) observer.next(value); },
    error: err => observer.error(err),
    complete: () => observer.complete()
  }).unsubscribe;
});

const debounceTime = (ms) => (source) => new Observable(observer => {
  let timer;
  return source.subscribe({
    next: value => {
      clearTimeout(timer);
      timer = setTimeout(() => observer.next(value), ms);
    },
    error: err => observer.error(err),
    complete: () => { clearTimeout(timer); observer.complete(); }
  }).unsubscribe;
});

const takeUntil = (notifier) => (source) => new Observable(observer => {
  const sub1 = source.subscribe(observer).unsubscribe;
  const sub2 = notifier.subscribe({ next: () => { observer.complete(); } }).unsubscribe;
  return () => { sub1(); sub2(); };
});

// 使用
const clicks$ = Observable.fromEvent(document, 'click');
const subscription = clicks$.pipe(
  map(e => ({ x: e.clientX, y: e.clientY })),
  filter(pos => pos.x > 100),
  debounceTime(300)
).subscribe(pos => console.log('Click at:', pos));

const input = document.getElementById('search');
const searchInput$ = Observable.fromEvent(input, 'input').pipe(
  map(e => e.target.value.trim()),
  filter(query => query.length >= 2),
  debounceTime(300),
  distinctUntilChanged(),
  switchMap(query => fromFetch(`/api/search?q=${query}`))
);

searchInput$.subscribe({
  next: results => renderSuggestions(results),
  error: err => showError(err)
});

class WebSocketManager extends EventEmitter {
  #ws;
  #reconnectDelay = 1000;
  
  connect(url) {
    this.#ws = new WebSocket(url);
    
    this.#ws.onmessage = (e) => {
      const message = JSON.parse(e.data);
      this.emit(message.type, message.payload);
      this.emit('*', message); // 通配符
    };
    
    this.#ws.onclose = () => {
      this.emit('disconnect');
      setTimeout(() => this.connect(url), this.#reconnectDelay);
      this.#reconnectDelay = Math.min(this.#reconnectDelay * 2, 30000);
    };
    
    this.#ws.onopen = () => {
      this.#reconnectDelay = 1000;
      this.emit('connect');
    };
  }
  
  send(type, payload) {
    this.#ws?.send(JSON.stringify({ type, payload }));
  }
}

class EventStore {
  #events = [];
  #subscribers = new Map();
  
  append(event) {
    const stored = {
      ...event,
      id: crypto.randomUUID(),
      timestamp: Date.now(),
      version: this.#events.length + 1
    };
    this.#events.push(stored);
    this.#notify(stored);
    return stored;
  }
  
  replay(fromVersion = 0) {
    return this.#events.filter(e => e.version >= fromVersion);
  }
  
  project(reducer, initialState) {
    return this.#events.reduce(reducer, initialState);
  }
  
  subscribe(eventType, handler) {
    if (!this.#subscribers.has(eventType)) this.#subscribers.set(eventType, new Set());
    this.#subscribers.get(eventType).add(handler);
    return () => this.#subscribers.get(eventType).delete(handler);
  }
  
  #notify(event) {
    this.#subscribers.get(event.type)?.forEach(h => h(event));
    this.#subscribers.get('*')?.forEach(h => h(event));
  }
}

模式	耦合度	适用场景	代表实现
观察者	高（直接引用）	组件内通信	MutationObserver
发布订阅	低（中介者）	跨模块通信	EventBus
事件流	低（管道）	复杂异步逻辑	RxJS
事件委托	—	DOM 事件优化	jQuery.on

事件流控制与观察者模式

事件流控制与观察者模式

目录

1. DOM 事件模型三阶段

2. EventEmitter 完整实现

3. 发布订阅 vs 观察者模式

4. 事件委托与高效绑定

5. 响应式事件流 (Observable)

6. 流操作符实现

7. 实战案例

实战案例 1：搜索框自动补全

实战案例 2：WebSocket 消息处理

实战案例 3：事件溯源（Event Sourcing）

8. 深度追问

Q1：passive 事件监听器如何提升滚动性能？

Q2：EventTarget 可以被继承吗？

Q3：如何避免事件监听器内存泄漏？

9. 总结表格