Architecture Pattern: Publish-subscribe Pattern

简介:

1. Brief                            

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

          

2. Role                            

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

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

      Message Broker or 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 - 被添加的消息实例
         * @param {Object.<{Number} GUIDofTimer, {Number} pause>} timer - 定时执行器
         */
        static addMessage(mq, message, timer){
                timer.pause = 1
                if (!(timer.pause = 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)
                timer.pause = 0
        }
        /* @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(sub.filter(message) && ++found)
                        sub.handler(message)
                    if (!found) unresponsedMq.push(message)
            }
            return unresponsedMq
        }
        static doUnsub(subs, pred){
            var sub, remainSubs = []
            while(sub = subs.shift()) if(pred());else
                remainSubs.push(sub)
            return remainSubs
        }
}

/**
 * @class
 * @private
 * @description 消息类
 */
class Message{
        constructor(topic, content, priority = 0){
                this.topic = topic
                this.content = content
                this.priority = priority
        }
}

/**
 * @class
 * @public
 * @description 消息中转站
 */
export default class MrB{
        constructor(during = 10){
            this.mq = []
            this.subs = []
            // dispatch message to subscribers, then recycle unhandled messages
            this.timer = {timer: setInterval(()=>{
                        if (this.timer.pause) return
                        this.mq = Utils.dispatch(this.mq, this.subs)
                    }, during)}
        }
        /* @method 订阅消息
         * @public
         * @param {DOMString|Function} filter - 消息过滤器
         * @param {Function} handler - 消息响应函数
         * @returns {DOMString} - 订阅编号,用于退订
         */
        sub(filter, handler){
                var guid = (+new Date()) + '' + 100000*Math.random()
                this.subs.push({guid: guid, filter: Utils.wrapFilter(filter), handler: handler})
                return guid
        }

        /* @method 退订 
         * @public 
         * @param {DOMString|Function} filter/guid - 消息过滤器 或 订阅编号
         * @param {Function} [handler] - 消息响应函数
         */
        unsub(filter/*or guid*/, handler){
                this.subs = Utils.doUnsub(this.subs, handler 
                                ? (sub)=>sub.filter.toString() === filter.toString() && sub.handler.toString() === handler.toString()
                                : (sub)=>sub.guid === filter.toString() || sub.filter.toString() === filter.toString())
        }

        /* @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, this.timer)
        }
}

6. Caution                            

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

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

 

7.  Idea                            

  在页面开发时我偏向于Component-Driven dev的开发模式,表面上是将页面切割为一个个功能独立的组件,本质上是将问题相关的依赖内聚,从而更好地识别、解决问题。而组件间的通信则是采用这种开 发模式所必定要考虑的另一个问题。解决方案有好多,但我觉得基本原则应该是:

  1. 由于组件是相互独立的松耦合结构,它们之间的通信不应该带来耦合度上扬的副作用;(若组件间通信是紧密的,应该考虑是否开发成子组件更为合适)

  2. 组件间通信的通道应该是配置的,这样才能灵活地对数据流作加工。(如:写日志、数据转换、类型转换等) 

  而采用Pub/Sub模式,利用消息作为组件间通信的数据载体,Message Broker负责信息的过滤和路由,实现消息在组件间的流转。另外可以通过定制Message Broker实现自定义组件通信通道,以AOP方式实现基础服务功能。

  而这种方式不可避免地会引入新问题:

  1. Message Broker作为消息中转站具有异步处理的特性,若需要同步执行,那么则需要引入另一种方式;

  2. 由于组件间松耦合,必须通过良好的日志记录方式来记录消息流转路径,否则无法debug。

相关实践学习
消息队列RocketMQ版:基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能,涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
目录
相关文章
|
设计模式 JavaScript 调度
JS案例:Observer Pattern(观察者模式)和Publisher-Subscriber Pattern(发布者/订阅者模式)
JS案例:Observer Pattern(观察者模式)和Publisher-Subscriber Pattern(发布者/订阅者模式)
136 0
JS案例:Observer Pattern(观察者模式)和Publisher-Subscriber Pattern(发布者/订阅者模式)
|
设计模式 安全 Java
设计模式|从Visitor Pattern说到Pattern Matching
在软件开发领域,我们每次遇到的问题可能都各不相同,有些是跟电商业务相关的,有些是跟底层数据结构相关的,而有些则可能重点在性能优化上。然而不管怎么样,我们在代码层面上解决问题的方法都有一定的共性。有没有人总结过这些共性呢?
设计模式|从Visitor Pattern说到Pattern Matching
|
Java
log4j中Pattern布局ConversionPattern详解
spring使用log4j,可以有2种方法。 1、在web.xml里不做任何配置。 log4j.properties放在classpath根目录下, 这时候生成的日志文件就没有相对路径,如果写相对路径,则会生成在安装tomcat的根路径下。 2、在web.xml设置。 &lt;context-param&gt;           &lt;param-name&gt;log4jConfi
1874 0