发布/订阅模式：解耦系统的强大设计模式

发布于：2025-07-02 ⋅ 阅读:(15) ⋅ 点赞:(0)

Hi，我是布兰妮甜 ！发布/订阅模式（Publish/Subscribe Pattern，简称 Pub/Sub）是一种消息传递模式，它允许发送者（发布者）将消息发送给多个接收者（订阅者），而无需知道这些接收者的具体信息。这种模式在现代软件开发中广泛应用，特别是在需要松耦合组件通信的场景中。

文章目录

一、发布/订阅模式概述

发布/订阅模式的核心思想是将消息的发送者和接收者解耦。在这种模式中：

  • 发布者（Publisher）：负责发布消息到特定频道或主题
  • 订阅者（Subscriber）：订阅感兴趣的频道或主题，接收相关消息
  • 消息代理（Broker）：负责管理频道和消息的路由（有时这个角色被隐含在实现中）

这种模式的主要优点包括：

  • 松耦合：发布者和订阅者不需要知道彼此的存在
  • 可扩展性：可以轻松添加新的发布者或订阅者
  • 灵活性：订阅者可以动态订阅或取消订阅主题

二、发布/订阅模式的实现方式

2.1 简单的事件发射器实现

class EventEmitter {
  constructor() {
    this.events = {};
  }

  // 订阅事件
  on(eventName, callback) {
    if (!this.events[eventName]) {
      this.events[eventName] = [];
    }
    this.events[eventName].push(callback);
  }

  // 发布事件
  emit(eventName, ...args) {
    const eventCallbacks = this.events[eventName];
    if (eventCallbacks) {
      eventCallbacks.forEach(callback => {
        callback(...args);
      });
    }
  }

  // 取消订阅
  off(eventName, callback) {
    const eventCallbacks = this.events[eventName];
    if (eventCallbacks) {
      this.events[eventName] = eventCallbacks.filter(cb => cb !== callback);
    }
  }

  // 一次性订阅
  once(eventName, callback) {
    const onceCallback = (...args) => {
      callback(...args);
      this.off(eventName, onceCallback);
    };
    this.on(eventName, onceCallback);
  }
}

// 使用示例
const emitter = new EventEmitter();

// 订阅者1
emitter.on('message', (msg) => {
  console.log(`订阅者1收到消息: ${msg}`);
});

// 订阅者2
emitter.on('message', (msg) => {
  console.log(`订阅者2收到消息: ${msg}`);
});

// 发布消息
emitter.emit('message', 'Hello, Pub/Sub!');
// 输出:
// 订阅者1收到消息: Hello, Pub/Sub!
// 订阅者2收到消息: Hello, Pub/Sub!

2.2 更复杂的主题订阅实现

class PubSub {
  constructor() {
    this.topics = {};
    this.hOP = this.topics.hasOwnProperty;
  }

  subscribe(topic, listener) {
    // 如果主题不存在，创建它
    if (!this.hOP.call(this.topics, topic)) this.topics[topic] = [];

    // 添加监听器到主题
    const index = this.topics[topic].push(listener) - 1;

    // 提供取消订阅的引用
    return {
      remove: () => {
        delete this.topics[topic][index];
      }
    };
  }

  publish(topic, info) {
    // 如果主题不存在或没有订阅者，直接返回
    if (!this.hOP.call(this.topics, topic)) return;

    // 通知所有订阅者
    this.topics[topic].forEach(item => {
      item(info !== undefined ? info : {});
    });
  }
}

// 使用示例
const pubsub = new PubSub();

// 订阅新闻主题
const subscription1 = pubsub.subscribe('news', (data) => {
  console.log('新闻订阅者收到:', data);
});

// 订阅天气主题
const subscription2 = pubsub.subscribe('weather', (data) => {
  console.log('天气订阅者收到:', data);
});

// 发布消息
pubsub.publish('news', { title: '新的发布/订阅文章发布', author: 'John Doe' });
pubsub.publish('weather', { forecast: '晴天', temperature: '25°C' });

// 取消订阅
subscription1.remove();

三、发布/订阅模式的实际应用场景

3.1 前端开发

// 在Vue.js中的事件总线
const EventBus = new Vue();

// 组件A - 发布事件
EventBus.$emit('user-logged-in', { username: 'john_doe' });

// 组件B - 订阅事件
EventBus.$on('user-logged-in', (user) => {
  console.log(`欢迎, ${user.username}`);
});

3.2 微服务架构

// 模拟微服务间的通信
class Microservice {
  constructor(name, broker) {
    this.name = name;
    this.broker = broker;
  }

  sendEvent(event, data) {
    this.broker.publish(event, { source: this.name, data });
  }

  listenTo(event, callback) {
    this.broker.subscribe(event, callback);
  }
}

const broker = new PubSub();

const userService = new Microservice('user-service', broker);
const orderService = new Microservice('order-service', broker);

// 订单服务监听用户创建事件
orderService.listenTo('user-created', (message) => {
  console.log(`[${orderService.name}] 收到用户创建事件:`, message);
  // 为新用户创建购物车等
});

// 用户服务发布事件
userService.sendEvent('user-created', { userId: 123, email: 'user@example.com' });

3.3 实时数据更新

// 股票价格更新示例
class StockTicker {
  constructor() {
    this.prices = {};
    this.subscribers = [];
    setInterval(() => this.updatePrices(), 1000);
  }

  subscribe(callback) {
    this.subscribers.push(callback);
    return () => {
      this.subscribers = this.subscribers.filter(sub => sub !== callback);
    };
  }

  updatePrices() {
    // 模拟价格变化
    const symbols = ['AAPL', 'GOOGL', 'MSFT'];
    symbols.forEach(symbol => {
      this.prices[symbol] = (Math.random() * 1000).toFixed(2);
    });
    
    // 通知所有订阅者
    this.subscribers.forEach(callback => callback(this.prices));
  }
}

const ticker = new StockTicker();

// 订阅价格更新
const unsubscribe = ticker.subscribe((prices) => {
  console.log('最新股价:', prices);
});

// 5秒后取消订阅
setTimeout(() => {
  unsubscribe();
  console.log('已取消订阅股票更新');
}, 5000);

四、发布/订阅模式的变体和高级特性

4.1 支持通配符的主题

class AdvancedPubSub {
  constructor() {
    this.topics = {};
  }

  subscribe(topic, callback) {
    const topicParts = topic.split('.');
    let currentLevel = this.topics;

    topicParts.forEach((part, index) => {
      if (!currentLevel[part]) {
        currentLevel[part] = {};
      }
      if (index === topicParts.length - 1) {
        currentLevel[part]._callbacks = currentLevel[part]._callbacks || [];
        currentLevel[part]._callbacks.push(callback);
      }
      currentLevel = currentLevel[part];
    });

    return {
      unsubscribe: () => {
        this._unsubscribe(topic, callback);
      }
    };
  }

  _unsubscribe(topic, callback) {
    const topicParts = topic.split('.');
    let currentLevel = this.topics;

    for (let i = 0; i < topicParts.length; i++) {
      const part = topicParts[i];
      if (!currentLevel[part]) return;
      if (i === topicParts.length - 1) {
        currentLevel[part]._callbacks = currentLevel[part]._callbacks.filter(cb => cb !== callback);
      }
      currentLevel = currentLevel[part];
    }
  }

  publish(topic, data) {
    const topicParts = topic.split('.');
    const matchedTopics = this._findMatchingTopics(topicParts);
    
    matchedTopics.forEach(t => {
      if (t._callbacks) {
        t._callbacks.forEach(callback => callback(data));
      }
    });
  }

  _findMatchingTopics(topicParts, level = this.topics, currentDepth = 0) {
    let results = [];
    const currentPart = topicParts[currentDepth];

    if (currentDepth === topicParts.length - 1) {
      if (currentPart === '*') {
        Object.keys(level).forEach(key => {
          if (key !== '_callbacks') {
            results = results.concat(this._findMatchingTopics(topicParts, level[key], currentDepth));
          }
        });
      } else if (level[currentPart]) {
        results.push(level[currentPart]);
      }
    } else {
      if (currentPart === '*') {
        Object.keys(level).forEach(key => {
          if (key !== '_callbacks') {
            results = results.concat(this._findMatchingTopics(topicParts, level[key], currentDepth + 1));
          }
        });
      } else if (level[currentPart]) {
        results = results.concat(this._findMatchingTopics(topicParts, level[currentPart], currentDepth + 1));
      }
    }

    return results;
  }
}

// 使用示例
const advancedPubSub = new AdvancedPubSub();

// 订阅特定主题
advancedPubSub.subscribe('news.tech', (data) => {
  console.log('收到科技新闻:', data);
});

// 订阅通配符主题
advancedPubSub.subscribe('news.*', (data) => {
  console.log('收到任何类型的新闻:', data);
});

// 发布消息
advancedPubSub.publish('news.tech', { title: '新的JavaScript框架发布' });
advancedPubSub.publish('news.sports', { title: '世界杯决赛结果' });

4.2 持久化订阅和历史消息

class PersistentPubSub {
  constructor() {
    this.topics = {};
    this.messageHistory = {};
    this.maxHistory = 100; // 每个主题保留的最大历史消息数
  }

  subscribe(topic, callback, receiveHistory = false) {
    if (!this.topics[topic]) {
      this.topics[topic] = [];
      this.messageHistory[topic] = [];
    }

    this.topics[topic].push(callback);

    // 如果需要，发送历史消息
    if (receiveHistory && this.messageHistory[topic].length > 0) {
      this.messageHistory[topic].forEach(msg => callback(msg));
    }

    return {
      unsubscribe: () => {
        this.topics[topic] = this.topics[topic].filter(cb => cb !== callback);
      }
    };
  }

  publish(topic, message) {
    if (!this.topics[topic]) {
      this.topics[topic] = [];
      this.messageHistory[topic] = [];
    }

    // 保存消息到历史
    this.messageHistory[topic].push(message);
    if (this.messageHistory[topic].length > this.maxHistory) {
      this.messageHistory[topic].shift();
    }

    // 通知所有订阅者
    this.topics[topic].forEach(callback => {
      callback(message);
    });
  }

  getHistory(topic) {
    return this.messageHistory[topic] || [];
  }
}

// 使用示例
const persistentPubSub = new PersistentPubSub();

// 发布一些初始消息
persistentPubSub.publish('system.updates', '系统启动');
persistentPubSub.publish('system.updates', '加载配置完成');

// 新订阅者请求历史消息
persistentPubSub.subscribe('system.updates', (msg) => {
  console.log('收到系统更新:', msg);
}, true); // 注意 receiveHistory 参数设置为 true

// 发布新消息
persistentPubSub.publish('system.updates', '用户登录');

五、发布/订阅模式的优缺点

优点：

  1. 松耦合：发布者和订阅者不需要知道对方的存在
  2. 可扩展性：可以轻松添加新的发布者或订阅者
  3. 灵活性：订阅者可以动态订阅或取消订阅
  4. 可重用性：消息代理的实现可以在不同项目中重用
  5. 易于测试：组件可以独立测试，因为它们不直接依赖其他组件

缺点：

  1. 调试困难：消息流可能变得复杂，难以跟踪
  2. 潜在的性能问题：如果有大量消息或订阅者，可能会影响性能
  3. 消息顺序问题：不能保证消息的接收顺序
  4. 消息丢失风险：大多数简单实现不保证消息的可靠传递
  5. 过度使用风险：可能导致系统设计过于依赖事件，使流程难以理解

六、发布/订阅模式与其他模式的比较

与观察者模式的区别

  • 观察者模式：观察者直接订阅特定主题（通常是单个主题），主题维护观察者列表
  • 发布/订阅模式：通过消息代理/频道解耦，发布者不知道订阅者的存在

与中介者模式的区别

  • 中介者模式：组件通过中介者直接通信，中介者知道所有组件
  • 发布/订阅模式：完全解耦，组件不知道彼此的存在

七、现代JavaScript中的发布/订阅

7.1 RxJS (Reactive Extensions)

import { Subject } from 'rxjs';

// 创建主题
const messageSubject = new Subject();

// 订阅者1
const subscription1 = messageSubject.subscribe({
  next: (msg) => console.log(`订阅者1: ${msg}`)
});

// 订阅者2
const subscription2 = messageSubject.subscribe({
  next: (msg) => console.log(`订阅者2: ${msg}`)
});

// 发布消息
messageSubject.next('Hello RxJS!');

// 取消订阅
subscription1.unsubscribe();

7.2 Redux中的事件总线

import { createStore } from 'redux';

// 简单的reducer
function reducer(state = {}, action) {
  switch (action.type) {
    case 'MESSAGE':
      console.log('收到消息:', action.payload);
      return state;
    default:
      return state;
  }
}

const store = createStore(reducer);

// 订阅状态变化
const unsubscribe = store.subscribe(() => {
  console.log('状态已更新:', store.getState());
});

// 发布消息 (dispatch action)
store.dispatch({ type: 'MESSAGE', payload: 'Hello Redux!' });

// 取消订阅
unsubscribe();

八、最佳实践和注意事项

  1. 合理设计主题结构：使用清晰的、有层次的主题命名（如 ‘user.created’、‘order.updated’）
  2. 避免过度使用：不是所有通信都需要通过发布/订阅，简单场景直接调用可能更合适
  3. 错误处理：确保订阅者中的错误不会影响整个系统
  4. 性能考虑：对于高频事件，考虑节流或防抖
  5. 内存管理：及时取消不再需要的订阅，避免内存泄漏
  6. 文档化事件：为系统使用的所有事件类型和数据结构维护文档
  7. 考虑持久化：对于关键消息，考虑实现持久化机制

九、总结

发布/订阅模式是构建松耦合、可扩展系统的强大工具。它在前端框架、微服务架构、实时系统等各种场景中都有广泛应用。JavaScript提供了多种实现方式，从简单的事件发射器到复杂的主题匹配系统，开发者可以根据项目需求选择合适的实现方式。

通过合理使用发布/订阅模式，可以创建出更加灵活、可维护的系统架构，但也要注意避免过度使用带来的复杂性和调试困难。理解其核心概念和适用场景，才能充分发挥这种模式的优势。

网站公告

今日签到

点亮在社区的每一天
去签到

热门文章

最新发布