2018/06/16 - 结合 Vue 源码谈谈发布-订阅模式

[lance10030]

结合Vue源码谈谈发布-订阅模式

最近的工作学习中接触到了发布-订阅模式。该思想编程中的应用也是很广泛的, 例如在 Vue中也大量使用了该设计模式,所以会结合Vue的源码和大家谈谈自己粗浅的理解.

发布订阅模式主要包含哪些内容呢?

1. 发布函数，发布的时候执行相应的回调
2. 订阅函数，添加订阅者,传入发布时要执行的函数,可能会携额外参数
3. 一个缓存订阅者以及订阅者的回调函数的列表
4. 取消订阅(需要分情况讨论)

这么看下来,其实就像 JavaScript 中的事件模型,我们在DOM节点上绑定事件函数,触发的时候执行就是应用了发布-订阅模式.

我们先按照上面的内容自己实现一个 Observer 对象如下:

//用于存储订阅的事件名称以及回调函数列表的键值对
function Observer() {
    this.cache = {}  
}

//key:订阅消息的类型的标识（名称），fn收到消息之后执行的回调函数
Observer.prototype.on = function (key,fn) {
    if(!this.cache[key]){
        this.cache[key]=[]
    }
    this.cache[key].push(fn)
}


//arguments 是发布消息时候携带的参数数组
Observer.prototype.emit = function (key) {
    if(this.cache[key]&&this.cache[key].length>0){
        var fns = this.cache[key]
    }
    for(let i=0;i<fns.length;i++){
        Array.prototype.shift.call(arguments)
        fns[i].apply(this,arguments)
    }
}
// remove 的时候需要注意，如果你直接传入一个匿名函数fn，那么你在remove的时候是无法找到这个函数并且把它移除的，变通方式是传入一个
//指向该函数的指针，而 订阅的时候存入的也是这个指针
Observer.prototype.remove = function (key,fn) {
    let fns = this.cache[key]
    if(!fns||fns.length===0){
        return
    }
    //如果没有传入fn，那么就是取消所有该事件的订阅
    if(!fn){
        fns=[]
    }else {
        fns.forEach((item,index)=>{
            if(item===fn){
                fns.splice(index,1)
            }
        })
    }
}


//example


var obj = new Observer()
obj.on('hello',function (a,b) {
    console.log(a,b)
})
obj.emit('hello',1,2)
//取消订阅事件的回调必须是具名函数
obj.on('test',fn1 =function () {
    console.log('fn1')
})
obj.on('test',fn2 = function () {
    console.log('fn2')
})
obj.remove('test',fn1)
obj.emit('test')

为什么会使用发布订阅模式呢? 它的优点在于:

1. 实现时间上的解耦(组件,模块之间的异步通讯)
2. 对象之间的解耦,交由发布订阅的对象管理对象之间的耦合关系.

发布-订阅模式在 Vue中的应用

1. Vue的实例方法中的应用:(当前版本:2.5.16)

• 文档传送门
• 源码传送门
• 源码节选 :(引入了flow.js用于静态类型检查)

// vm.$on
export function eventsMixin (Vue: Class<Component>) {
    const hookRE = /^hook:/
    //参数类型为字符串或者字符串组成的数组
    Vue.prototype.$on = function (event: string | Array<string>, fn: Function): Component {
        const vm: Component = this
        // 传入类型为数组
        if (Array.isArray(event)) {
            for (let i = 0, l = event.length; i < l; i++) {
                this.$on(event[i], fn)
                //递归并传入相应的回调
            }
        } else {
        //
            (vm._events[event] || (vm._events[event] = [])).push(fn)
            // optimize hook:event cost by using a boolean flag marked at registration
            // instead of a hash lookup
            if (hookRE.test(event)) {
                vm._hasHookEvent = true
            }
        }
        return vm
    }


// vm.$emit

 Vue.prototype.$emit = function (event: string): Component {
    const vm: Component = this
    if (process.env.NODE_ENV !== 'production') {
      const lowerCaseEvent = event.toLowerCase()
      if (lowerCaseEvent !== event && vm._events[lowerCaseEvent]) {
        tip(
          `Event "${lowerCaseEvent}" is emitted in component ` +
          `${formatComponentName(vm)} but the handler is registered for "${event}". ` +
          `Note that HTML attributes are case-insensitive and you cannot use ` +
          `v-on to listen to camelCase events when using in-DOM templates. ` +
          `You should probably use "${hyphenate(event)}" instead of "${event}".`
        )
      }
    }
    let cbs = vm._events[event]
    if (cbs) {
      cbs = cbs.length > 1 ? toArray(cbs) : cbs
      const args = toArray(arguments, 1)
      for (let i = 0, l = cbs.length; i < l; i++) {
        try {
          cbs[i].apply(vm, args)// 执行之前传入的回调
        } catch (e) {
          handleError(e, vm, `event handler for "${event}"`)
        }
      }
    }
    return vm
  }

Vue中还实现了vm.$once (监听一次);以及vm.$off (取消订阅) ,大家可以在同一文件中看一下是如何实现的.

2. Vue数据更新机制中的应用

• observer每个对象的属性，添加到订阅者容器Dependency(Dep)中，当数据发生变化的时候发出notice通知。
• Watcher：某个属性数据的监听者/订阅者，一旦数据有变化，它会通知指令(directive)重新编译模板并渲染UI
• 部分源码如下: 源码传送门-observer

export class Observer {
  value: any;
  dep: Dep;
  vmCount: number; // number of vms that has this object as root $data

  constructor (value: any) {
    this.value = value
    this.dep = new Dep()
    this.vmCount = 0
    def(value, '__ob__', this)
    if (Array.isArray(value)) {
      const augment = hasProto
        ? protoAugment
        : copyAugment
      augment(value, arrayMethods, arrayKeys)
      this.observeArray(value)
    } else {
      this.walk(value)
    }
  }

  /**
   * Walk through each property and convert them into
   * getter/setters. This method should only be called when
   * value type is Object.
   */
   // 属性为对象的时候,observe 对象的属性
  walk (obj: Object) {
    const keys = Object.keys(obj)
    for (let i = 0; i < keys.length; i++) {
      defineReactive(obj, keys[i])
    }
  }

  /**
   * Observe a list of Array items.
   */
  observeArray (items: Array<any>) {
    for (let i = 0, l = items.length; i < l; i++) {
      observe(items[i])
    }
  }
}

• Dep对象: 订阅者容器，负责维护watcher源码传送门

export default class Dep {
  static target: ?Watcher;
  id: number;
  subs: Array<Watcher>;

  constructor () {
    this.id = uid++
    this.subs = []   //存储订阅者 
  }
  // 添加watcher
  addSub (sub: Watcher) {
    this.subs.push(sub)
  }
 // 移除
  removeSub (sub: Watcher) {
    remove(this.subs, sub)
  }

  depend () {
    if (Dep.target) {
      Dep.target.addDep(this)
    }
  }
 // 变更通知
  notify () {
    // stabilize the subscriber list first
    const subs = this.subs.slice()
    for (let i = 0, l = subs.length; i < l; i++) {
      subs[i].update()
    }
  }
}

工作中小应用举例

1. 场景: 基于wepy的小程序. 由于项目本身不是足够的复杂到要使用提供的 redux进行状态管理.但是在不同的组件(不限于父子组件)之间,存在相关联的异步操作.所以在wepy对象上挂载了一个本文最开始实现的Observer对象.作为部分组件之间通信的总线机制:

wepy.$bus = new Observer()
// 然后就可以在不同的模块和组件中订阅和发布消息了

要注意的点

当然,发布-订阅模式也是有缺点的.

1. 创建订阅者本身会消耗内存,订阅消息后,也许,永远也不会有发布,而订阅者始终存在内存中.
2. 对象之间解耦的同时,他们的关系也会被深埋在代码背后,这会造成一定的维护成本.

当然设计模式的存在是帮助我们解决特定场景的问题的,学会在正确的场景中使用才是最重要的.

广而告之

本文发布于薄荷前端周刊，欢迎Watch & Star ★，转载请注明出处。

欢迎讨论，点个赞再走吧 ｡◕‿◕｡ ～

[sarazhang123]

提个小小的bug，fns = [] 并不能清除所有事件的订阅，需要用this.cache[key] = []来清除所有事件的callback。

  //如果没有传入fn，那么就是取消所有该事件的订阅
   if(!fn){
        fns=[]
    }else {
        fns.forEach((item,index)=>{
            if(item===fn){
                fns.splice(index,1)
            }
        })
    }