概念
发布-订阅模式又称为观察者模式，它定义的是一种一对多的依赖关系，当一个状态发生改变的时候，所有以来这个状态的对象都会得到通知。
生活中的发布-订阅模式
上面事发布-订阅模式的一个比较正式的解释，可能这个解释不大好理解。所以我们通过实际生活中的例子来理解。
比如看中了一套房子，等到去了售楼处的说以后才被告知房子已经售罄了。但是售楼小姐告知，将来会有尾盘推出。具体什么时候推出，目前没人知道。
但是买家又不想频繁的跑，于是就把自己的电话号码登记在售楼处，在登记的花名册上有很多类似的买家。售楼小姐答应买家，新的房源一出来就一一通知买家。
所以上面就是一个发布订阅模式的简单例子。购房者（订阅者）订阅房源信息，售楼处（发布者）发布新房源消息给购房者（订阅者），购房者（订阅者）接收到消息后作出相应的反应。
适用性
- 发布订阅模式可以广泛的应用于异步编程中。
- 发布订阅模式可以取代对象之间的硬编码通知机制。
典型实现例子
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) {
// 购买者a
console.log(`价格是：${price}`)
console.log(`面积大小：${squareMeter}`)
})
salesOffices.listen(function(price, squareMeter) {
// 购买者b
console.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/setters
const getter = property && property.get
const 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) : val
if (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 ob2."中转站"
在第一步数据劫持的时候，数据的获取或者修改的时候，都会做出对应的操作。这些操作的目的很简单，就是“通知”到“中转站”。这个“中转站”主要就是对数据的变更起通知作用以及存放依赖这些数据的“地方”。
这个"中转站"就是由"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: ?Watcher
id: number
subs: 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 first
const 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: Component
expression: string
cb: Function
id: number
deep: boolean
user: boolean
lazy: boolean
sync: boolean
dirty: boolean
active: boolean
deps: Array<Dep>
newDeps: Array<Dep>
depIds: ISet
newDepIds: ISet
getter: Function
value: any
constructor(
vm: Component,
expOrFn: string | Function,
cb: Function,
options?: Object
) {
this.vm = vm
/*_watchers存放订阅者实例*/
vm._watchers.push(this)
// options
if (options) {
this.deep = !!options.deep
this.user = !!options.user
this.lazy = !!options.lazy
this.sync = !!options.sync
} else {
this.deep = this.user = this.lazy = this.sync = false
}
this.cb = cb
this.id = ++uid // uid for batching
this.active = true
this.dirty = this.lazy // for lazy watchers
this.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 value
const 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.id
if (!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.length
while (i--) {
const dep = this.deps[i]
if (!this.newDepIds.has(dep.id)) {
dep.removeSub(this)
}
}
let tmp = this.depIds
this.depIds = this.newDepIds
this.newDepIds = tmp
this.newDepIds.clear()
tmp = this.deps
this.deps = this.newDeps
this.newDeps = tmp
this.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 value
const 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.length
while (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.length
while (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 = 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)
}
return vm
}
Vue.prototype.$emit = function(event: string): Component {
const vm: Component = this
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++) {
cbs[i].apply(vm, args)
}
}
return vm
}总结
本文通过对 vue 相关源码的学习，了解了发布订阅模式（观察者模式）的概念和应用。还了解了该模式的 一些优缺点：
- 时间上的解耦，对象之间的解耦。
- 创建订阅者本身会消耗一定的时间和内存，并且订阅者订阅一个消息后，该消息一直不发生的话，那么该订阅者 会一直存在在内存中
感谢
《javascript 设计模式与开发实践》