概念

发布-订阅模式又称为观察者模式，它定义的是一种一对多的依赖关系，当一个状态发生改变的时候，所有以来这个状态的对象都会得到通知。

生活中的发布-订阅模式

上面事发布-订阅模式的一个比较正式的解释，可能这个解释不大好理解。所以我们通过实际生活中的例子来理解。

比如看中了一套房子，等到去了售楼处的说以后才被告知房子已经售罄了。但是售楼小姐告知，将来会有尾盘推出。具体什么时候推出，目前没人知道。

但是买家又不想频繁的跑，于是就把自己的电话号码登记在售楼处，在登记的花名册上有很多类似的买家。售楼小姐答应买家，新的房源一出来就一一通知买家。

所以上面就是一个发布订阅模式的简单例子。购房者（订阅者）订阅房源信息，售楼处（发布者）发布新房源消息给购房者（订阅者），购房者（订阅者）接收到消息后作出相应的反应。

适用性

• 发布订阅模式可以广泛的应用于异步编程中。
• 发布订阅模式可以取代对象之间的硬编码通知机制。

典型实现例子

1、售楼处的例子

一步步实现发布订阅模式：

• 首先指定好谁充当发布者（售楼处）
• 然后给发布者添加一个缓存列表，用语存放回调函数，以便通知订阅者（售楼处花名册）。
• 最后发布消息的时候，发布者会遍历这个缓存列表，依次触发里面存放的订阅者的回调函数。

let salesOffices = {} // 售楼处
salesOffices.books = [] // 缓存列表，存放订阅者的回调函数。
// 增加订阅者
salesOffices.listen = function(fn) {this.books.push(fn) // 订阅的消息添加近缓存列表里面
}
salesOffices.trigger = function() {// 发布消息for (let i = 0, fn; (fn = salesOffices.books[i++]); ) {fn.apply(this, arguments) // arguments 是发布消息的时候带上的参数}
}salesOffices.listen(function(price, squareMeter) {// 购买者aconsole.log(`价格是：${price}`)console.log(`面积大小：${squareMeter}`)
})
salesOffices.listen(function(price, squareMeter) {// 购买者bconsole.log(`价格是：${price}`)console.log(`面积大小：${squareMeter}`)
})salesOffices.trigger(2000000, 88)
salesOffices.trigger(3000000, 128)

上面实现了一个最简单的发布订阅模式。肯定还有很多问题的，例如订阅者只订阅了某一个消息，但是上面会把所有消息发给每一个订阅者。所以还得通过其他的方式让订阅者只订阅自己感兴趣的消息。

2、vue 对发布订阅模式的使用

我们都知道 Vue 有个最显著的特性，便是侵入性不是很强的响应式系统。这个特性就是对发布订阅模式非常好的应用。我们接下来就来看看这个特性是怎么应用的。

vue 的数据初始化：

var v = new Vue({data() {return {a: 'hello'}}
})

这个初始化的代码的背后包含着发布订阅模式的思想，接下来看看官网的一个图

接下来就是网友的一个图：@xuqiang521

1. 数据劫持

从上图可以看到，数据劫持的核心方法就是使用Object.defineProperty把属性转化成getter/setter。（因为这个是 ES5 中的方法，所以这也是 Vue 不支持 ie8 及以下浏览器的原因之一。）在数据传递变更的时候，会进入到我们封装的Dep和Watcher中进行处理。

1.1 遍历劫持

数据不紧紧是基本类型的数据，也有可能是对象或者数组。基本类型的数据和对象的处理起来比较简单。

walk(obj) {const keys = Object.keys(obj)for (let i = 0; i < keys.length; ++i) {defineReactive(obj, keys[i], obj[keys[i]])}
}

核心的劫持相关函数以及属性的订阅和发布

/*** Define a reactive property on an Object.*/
export function defineReactive(obj: Object,key: string,val: any,customSetter?: Function
) {/*在闭包中定义一个dep对象*/const dep = new Dep()const property = Object.getOwnPropertyDescriptor(obj, key)if (property && property.configurable === false) {return}/*如果之前该对象已经预设了getter以及setter函数则将其取出来，新定义的getter/setter中会将其执行，保证不会覆盖之前已经定义的getter/setter。*/// cater for pre-defined getter/settersconst getter = property && property.getconst setter = property && property.set/*对象的子对象递归进行observe并返回子节点的Observer对象*/let childOb = observe(val)Object.defineProperty(obj, key, {enumerable: true,configurable: true,get: function reactiveGetter() {/*如果原本对象拥有getter方法则执行*/const value = getter ? getter.call(obj) : valif (Dep.target) {/*进行依赖收集*/dep.depend()if (childOb) {/*子对象进行依赖收集，其实就是将同一个watcher观察者实例放进了两个depend中，一个是正在本身闭包中的depend，另一个是子元素的depend*/childOb.dep.depend()}if (Array.isArray(value)) {/*是数组则需要对每一个成员都进行依赖收集，如果数组的成员还是数组，则递归。*/dependArray(value)}}return value},set: function reactiveSetter(newVal) {/*通过getter方法获取当前值，与新值进行比较，一致则不需要执行下面的操作*/const value = getter ? getter.call(obj) : val/* eslint-disable no-self-compare */if (newVal === value || (newVal !== newVal && value !== value)) {return}/* eslint-enable no-self-compare */if (process.env.NODE_ENV !== 'production' && customSetter) {customSetter()}if (setter) {/*如果原本对象拥有setter方法则执行setter*/setter.call(obj, newVal)} else {val = newVal}/*新的值需要重新进行observe，保证数据响应式*/childOb = observe(newVal)/*dep对象通知所有的观察者*/dep.notify()}})
}

最开始在初始化的时候是对 data 里面的数据就开始劫持监听了。初始化的时候就调用了observe方法

/*** Attempt to create an observer instance for a value,* returns the new observer if successfully observed,* or the existing observer if the value already has one.*/
/*尝试创建一个Observer实例（__ob__），如果成功创建Observer实例则返回新的Observer实例，如果已有Observer实例则返回现有的Observer实例。*/
export function observe(value: any, asRootData: ?boolean): Observer | void {/*判断是否是一个对象*/if (!isObject(value)) {return}let ob: Observer | void/*这里用__ob__这个属性来判断是否已经有Observer实例，如果没有Observer实例则会新建一个Observer实例并赋值给__ob__这个属性，如果已有Observer实例则直接返回该Observer实例*/if (hasOwn(value, '__ob__') && value.__ob__ instanceof Observer) {ob = value.__ob__} else if (/*这里的判断是为了确保value是单纯的对象，而不是函数或者是Regexp等情况。*/observerState.shouldConvert &&!isServerRendering() &&(Array.isArray(value) || isPlainObject(value)) &&Object.isExtensible(value) &&!value._isVue) {ob = new Observer(value)}if (asRootData && ob) {/*如果是根数据则计数，后面Observer中的observe的asRootData非true*/ob.vmCount++}return ob
}

1.2 返回值

上面的数据observe之后返回的就是一个 Observer 的实例

ob = new Observer(value)return ob

2."中转站"

在第一步数据劫持的时候，数据的获取或者修改的时候，都会做出对应的操作。这些操作的目的很简单，就是“通知”到“中转站”。这个“中转站”主要就是对数据的变更起通知作用以及存放依赖这些数据的“地方”。

这个"中转站"就是由"Dep"和“Watcher” 类构成的。每个被劫持的数据都会产生一个这样的“中转站”

2.1 Dep

Dep，全名 Dependency，从名字我们也能大概看出 Dep 类是用来做依赖收集的，但是也有通知对应的订阅者的作用 ，让它执行自己的操作，具体怎么收集呢？

/*** A dep is an observable that can have multiple* directives subscribing to it.*/
export default class Dep {static target: ?Watcherid: numbersubs: Array<Watcher>constructor() {this.id = uid++this.subs = []}/*添加一个观察者对象*/addSub(sub: Watcher) {this.subs.push(sub)}/*移除一个观察者对象*/removeSub(sub: Watcher) {remove(this.subs, sub)}/*依赖收集，当存在Dep.target的时候添加观察者对象*/depend() {if (Dep.target) {Dep.target.addDep(this)}}/*通知所有订阅者*/notify() {// stabilize the subscriber list firstconst subs = this.subs.slice()for (let i = 0, l = subs.length; i < l; i++) {subs[i].update()}}
}// the current target watcher being evaluated.
// this is globally unique because there could be only one
// watcher being evaluated at any time.
Dep.target = null
/*依赖收集完需要将Dep.target设为null，防止后面重复添加依赖。*/
const targetStack = []
export function pushTarget(_target: Watcher) {if (Dep.target) targetStack.push(Dep.target)// 改变目标指向Dep.target = _target
}export function popTarget() {// 删除当前目标，重算指向Dep.target = targetStack.pop()
}

代码很简短，但它做的事情却很重要

• 定义 subs 数组，用来收集订阅者 Watcher
• 当劫持到数据变更的时候，通知订阅者 Watcher 进行 update 操作

2.2 Watcher

Watcher 就是订阅者（观察者）。 主要的作用就是就是订阅 Dep(每个属性都会有一个 dep)，当 Dep 发出消息传递（notify）的时候，所以订阅着 Dep 的 Watchers 会进行自己的 update 操作。

export default class Watcher {vm: Componentexpression: stringcb: Functionid: numberdeep: booleanuser: booleanlazy: booleansync: booleandirty: booleanactive: booleandeps: Array<Dep>newDeps: Array<Dep>depIds: ISetnewDepIds: ISetgetter: Functionvalue: anyconstructor(vm: Component,expOrFn: string | Function,cb: Function,options?: Object) {this.vm = vm/*_watchers存放订阅者实例*/vm._watchers.push(this)// optionsif (options) {this.deep = !!options.deepthis.user = !!options.userthis.lazy = !!options.lazythis.sync = !!options.sync} else {this.deep = this.user = this.lazy = this.sync = false}this.cb = cbthis.id = ++uid // uid for batchingthis.active = truethis.dirty = this.lazy // for lazy watchersthis.deps = []this.newDeps = []this.depIds = new Set()this.newDepIds = new Set()this.expression =process.env.NODE_ENV !== 'production' ? expOrFn.toString() : ''// parse expression for getter/*把表达式expOrFn解析成getter*/if (typeof expOrFn === 'function') {this.getter = expOrFn} else {this.getter = parsePath(expOrFn)if (!this.getter) {this.getter = function() {}process.env.NODE_ENV !== 'production' &&warn(`Failed watching path: "${expOrFn}" ` +'Watcher only accepts simple dot-delimited paths. ' +'For full control, use a function instead.',vm)}}this.value = this.lazy ? undefined : this.get()}/*** Evaluate the getter, and re-collect dependencies.*//*获得getter的值并且重新进行依赖收集*/get() {/*将自身watcher观察者实例设置给Dep.target，用以依赖收集。*/pushTarget(this)let valueconst vm = this.vm/*执行了getter操作，看似执行了渲染操作，其实是执行了依赖收集。在将Dep.target设置为自生观察者实例以后，执行getter操作。譬如说现在的的data中可能有a、b、c三个数据，getter渲染需要依赖a跟c，那么在执行getter的时候就会触发a跟c两个数据的getter函数，在getter函数中即可判断Dep.target是否存在然后完成依赖收集，将该观察者对象放入闭包中的Dep的subs中去。*/if (this.user) {try {value = this.getter.call(vm, vm)} catch (e) {handleError(e, vm, `getter for watcher "${this.expression}"`)}} else {value = this.getter.call(vm, vm)}// "touch" every property so they are all tracked as// dependencies for deep watching/*如果存在deep，则触发每个深层对象的依赖，追踪其变化*/if (this.deep) {/*递归每一个对象或者数组，触发它们的getter，使得对象或数组的每一个成员都被依赖收集，形成一个“深（deep）”依赖关系*/traverse(value)}/*将观察者实例从target栈中取出并设置给Dep.target*/popTarget()this.cleanupDeps()return value}/*** Add a dependency to this directive.*//*添加一个依赖关系到Deps集合中*/addDep(dep: Dep) {const id = dep.idif (!this.newDepIds.has(id)) {this.newDepIds.add(id)this.newDeps.push(dep)if (!this.depIds.has(id)) {dep.addSub(this)}}}/*** Clean up for dependency collection.*//*清理依赖收集*/cleanupDeps() {/*移除所有观察者对象*/let i = this.deps.lengthwhile (i--) {const dep = this.deps[i]if (!this.newDepIds.has(dep.id)) {dep.removeSub(this)}}let tmp = this.depIdsthis.depIds = this.newDepIdsthis.newDepIds = tmpthis.newDepIds.clear()tmp = this.depsthis.deps = this.newDepsthis.newDeps = tmpthis.newDeps.length = 0}/*** Subscriber interface.* Will be called when a dependency changes.*//*调度者接口，当依赖发生改变的时候进行回调。*/update() {/* istanbul ignore else */if (this.lazy) {this.dirty = true} else if (this.sync) {/*同步则执行run直接渲染视图*/this.run()} else {/*异步推送到观察者队列中，由调度者调用。*/queueWatcher(this)}}/*** Scheduler job interface.* Will be called by the scheduler.*//*调度者工作接口，将被调度者回调。*/run() {if (this.active) {const value = this.get()if (value !== this.value ||// Deep watchers and watchers on Object/Arrays should fire even// when the value is the same, because the value may// have mutated./*即便值相同，拥有Deep属性的观察者以及在对象／数组上的观察者应该被触发更新，因为它们的值可能发生改变。*/isObject(value) ||this.deep) {// set new valueconst oldValue = this.value/*设置新的值*/this.value = value/*触发回调渲染视图*/if (this.user) {try {this.cb.call(this.vm, value, oldValue)} catch (e) {handleError(e, this.vm, `callback for watcher "${this.expression}"`)}} else {this.cb.call(this.vm, value, oldValue)}}}}/*** Evaluate the value of the watcher.* This only gets called for lazy watchers.*//*获取观察者的值*/evaluate() {this.value = this.get()this.dirty = false}/*** Depend on all deps collected by this watcher.*//*收集该watcher的所有deps依赖*/depend() {let i = this.deps.lengthwhile (i--) {this.deps[i].depend()}}/*** Remove self from all dependencies' subscriber list.*//*将自身从所有依赖收集订阅列表删除*/teardown() {if (this.active) {// remove self from vm's watcher list// this is a somewhat expensive operation so we skip it// if the vm is being destroyed./*从vm实例的观察者列表中将自身移除，由于该操作比较耗费资源，所以如果vm实例正在被销毁则跳过该步骤。*/if (!this.vm._isBeingDestroyed) {remove(this.vm._watchers, this)}let i = this.deps.lengthwhile (i--) {this.deps[i].removeSub(this)}this.active = false}}
}

通过上面对 vue 的响应系统的 学习，就可以了解到这个发布订阅模式就是这样的：

• Dep 负责收集所有相关的的订阅者 Watcher ，具体谁不用管，具体有多少也不用管，只需要根据 target 指向的计算去收集订阅其消息的 Watcher 即可，然后做好消息发布 notify 即可。
• Watcher 负责订阅 Dep ，并在订阅的时候让 Dep 进行收集，接收到 Dep 发布的消息时，做好其 update 操作即可。

3、vue 中更多的应用

vue 中还有个组件之间的时间传递也是用到了发布订阅模式。
$emit 负责发布消息， $on 负责消费消息（执行 cbs 里面的事件）

Vue.prototype.$on = function(event: string | Array<string>,fn: Function
): Component {const vm: Component = thisif (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)}return vm
}Vue.prototype.$emit = function(event: string): Component {const vm: Component = thislet cbs = vm._events[event]if (cbs) {cbs = cbs.length > 1 ? toArray(cbs) : cbsconst args = toArray(arguments, 1)for (let i = 0, l = cbs.length; i < l; i++) {cbs[i].apply(vm, args)}}return vm
}

总结

本文通过对 vue 相关源码的学习，了解了发布订阅模式（观察者模式）的概念和应用。还了解了该模式的 一些优缺点：

• 时间上的解耦，对象之间的解耦。
• 创建订阅者本身会消耗一定的时间和内存，并且订阅者订阅一个消息后，该消息一直不发生的话，那么该订阅者 会一直存在在内存中

感谢

从源码角度再看数据绑定

《javascript 设计模式与开发实践》