转载

Architecture Pattern: Publish-subscribe Pattern

1. Brief

一直对Observer Pattern和Pub/Sub Pattern有所混淆，下面打算通过这两篇Blog来梳理这两种模式。若有纰漏请大家指正。

2. Role

Publisher：消息发布者，组装原始消息实体并触发消息传递的主体。

Subscriber：消息订阅者，接收消息实体并作出响应的主体。

Message Brokeror Event Bus ：消息发布者与消息订阅者间的媒介，内含消息过滤和消息路由的功能，并可通过内置的消息队列(message queue)现实优先级处理。

其中过滤功能又细分为topic-based和content-based两种类型

topic-based就是为消息主题建立独立通道，订阅特定主题的订阅者将通过对应的通道接收消息；

content-based就是以消息内容为处理源，订阅者仅接收消息内容与目标关键词匹配的消息。

此外还可以采用hybird=topic-based和content-based的方式。

3. Advantages

1. Loose coupling

由于引入Message Broker来处理消息过滤和路由等功能，而Subscriber又是通过Messsage Broker来订阅消息，从而实现Publisher和Subscriber的低耦合。这种低耦合体现在两个方面。

空间：Publisher和Subscriber可运行在两个不同的进程，甚至是机器上；

时间：Publisher和Subscriber不必同时运行，通过Message Broker作为消息中转站暂存消息( Store and forward )并实现消息异步处理。

2. Scalability

由于Publisher和Subscriber在空间和时间上松耦合，那么我们就可以通过增加进程/机器的方式来提高处理能力。

4. Disadvantages —— Semantic Coupling

"The most insidious kind of coupling occurs when one module makes use, not of some syntactic element of another module, but of some semantic knowledge of another module's inner workings" —— chapter 5 of Code Complete

语义耦合(Semantic Coupling)，是一种隐晦的耦合类型，导致代码重构、调试、修改复杂度急剧增加的主要原因。

类型如下：

1. 操作顺序耦合：使用一个对象，需要先调用Init()，之后才能调用DoAnything()。这种顺序耦合，即使在文档中remark也是极为不优雅的做法；

2. 全局参数传递：模块A修改了某个全局参数g_val，模块B读取该值。模块B必须知道模块A已经对该参数赋值；

3. 业务封装不够紧密：模块A向模块B传一个参数，模块B根据该参数选择对应的操作。模块A必须知道与业务相关的所有的操作类型。对于模块A，仅传递模块A自身可以理解的语义，或者通俗的概念作为参数，而不是被封装的业务相关的参数；

4. 超越接口的数据类型约定：模块A向模块B传递一个接口的指针，模块B将其强制转换为派生类的指针。当模块B知道该接口的实际类型时，封装已经被破坏了。非相关模块只能对接口操作，而不应对接口之外的职责进行约定。

public interface Customer{} public class VIP : Customer{   public void Serve(){} }  public static void main(String[] args){   Customer customer = new VIP();   Serve(customer); } public static void Serve(Customer customer){   VIP vip = (VIP)customer;   vip.Serve(); }

5. Simple implementation

/**  * @class  * @private  * @description 帮助类  */ class Utils{         /* @method 对用户输入的filter进行加工           * @static          * @package          * @param {DOMString|Function} origFilter - 用户输出的filter          * @returns {Function}          * @exception          */         static wrapFilter(origFilter){             var type = typeof origFilter             if ('string' === type) return message => RegExp(origFilter, 'i').test(message.topic)             else if ('function' === type) return origFilter             throw Error('the type of argument(0) is wrong, which accepts string or function instance only!')         }         /* @method 添加消息到消息队列          * @static          * @package          * @param {Array.<Message>} mq - 消息队列          * @param {Message} message - 被添加的消息实例          */         static addMessage(mq, message){                 if (!mq.length) return mq.push(message)                 for(var i = 0, inserted = 0, m; !inserted && (m = mq[i]); ++i)                      if(inserted = m.priority < message.priority)                         mq.splice(i, 0, message)                 if(!inserted) mq.push(message)         }         /* @method 分发消息给订阅者          * @static          * @package          * @param {Array.<Message>} mq - 消息队列          * @param {Array} subs - 订阅者池           * @returns {Array.<Message>} - 未被响应的消息回收队列          */         static dispatch(mq, subs){             var message, unresponsedMq = []             while(message = mq.shift()){                     let found = 0                     for(let sub of subs) if(found = sub.filter(message)){                         sub.handler(message)                         break                     }                     if (!found) unresponsedMq.push(message)             }             return unresponsedMq         } }  /**  * @class  * @private  * @description 消息类  */ class Message{         constructor(topic, content, priority = 0){                 this.topic = topic                 this.content = content                 this.priority = priority         } }  /**  * @class  * @public  * @description 消息中转站Mr. B(roker)  */ export default class MrB{         constructor(){             this.mq = []             this.subs = []         }         /* @method 订阅消息          * @public          * @param {DOMString|Function} filter - 消息过滤器          * @param {Function} handler - 消息响应函数          */         sub(filter, handler){                 this.subs.push({filter: Utils.wrapFilter(filter), handler: handler})         }          /* @method 发布消息          * @public          * @param {DOMString|Object|Message} topic - 消息主题 或 消息实体          * @param {DOMString} [content=''] - 消息内容          * @param {Number} [priority=0] - 消息优先级          */         pub(topic, content = '', priority = 0){                 var message                 if(1 === arguments.length) message = (topic.priority = topic.priority || 0, topic)                 else message = new Message(topic, content, priority)                                  // push message to mq                 Utils.addMessage(this.mq, message)                 // dispatch message to subscribers, then recycle unhandled messages                 this.mq = Utils.dispatch(this.mq, this.subs)         } }

6. Caution

1. Pub/Sub模式是Messaging模式的一种，而Messaging模式是一种基于网络的架构模式(network-oriented architectural pattern)，也就是说是以跨进程通信为应用范围；而Observer模式则是基于对象事件的设计模式(object-event oriented pattern)，并且其应用范围是单进程内的。

2. Pub/Sub模式适用于非实时处理;