手撕源码系列 —— 函子 + 观察者模式 + 状态 = Promise
作者: 图恩分类: 编程开发阅读: 639发布时间： 2020-04-14 02:25:15
前言
前段时间太忙，隔了快一个月没写博客，但是 Promise 其实很早之前就已经总结了一波如何实现，但是那个时候纯粹是为了实现而实现，没有去细品其中的一些巧妙设计，直到最近在进行函数式编程相关的知识学习时，无意中在查阅资料的时候发现，Promise 和 Functor 居然有着千丝万缕的关系，这让我决定要重新审视一下自己对 Promise 的认知，于是便有了这篇“老酒新装”的博客。
前置知识
想要完全阅读并理解这篇博客，我粗略地估算了一下，大概需要以下的一些前置知识，不了解的同学可以自行先去学习一下：
递归的思想
ES6 和 Typescript 的基础认知
函数式编程中的函子(Functor)和“函数是一等公民”思想
设计模式中的观察者模式
分步实现 Promise
我们可以将实现一个 Promise 比作盖一栋楼，通过拆分每一步并解决、理解和记忆，达到很快就能理解它的实现的目的。
一、打下 Promise 的地基
由于我们常常使用 Promise 配合 async/await 来实现异步编程，所以先回想一下如何最最基本地使用 Promise
new Promise(resolve => {...})
.then(onFulfilled)
.then(onFulfilled)
根据上面我们回忆中的最最基本的 Promise 我们可以写出以下实现：
class Promise {
constructor(executor) {
const resolve = () => {}
executor(resolve)
}
then = onFulfilled => new Promise(...)
}
好的，那我们的第一步到这里就结束了，是不是十分简单，完全不需要什么成本就能理解并记忆下来。
二、加入原材料 函子 和 观察者模式
这一步，我们开始往里加东西了。对于加的东西，我们也必须了解是什么，所以先来看下两个原材料的基本概念。
函子
由于在 前置知识 里提到，相信大家已经对它有所了解，一个基本的函子我们可以写成以下实现：
class Functor {
static of = value => new Functor(this.value)
constructor(value) {
this.value = value
}
map = fn => Functor.of(fn(this.value))
}
为了方便映射到 Promise 的实现中，改为以下写法：
class Functor {
constructor(value) {
this.value = value
}
map = fn => new Functor(fn(this.value))
}
然后结合到函子的一些特性：
const plus = x + y => x + y
const plusOne = x => plus(x, 1)
new Functor(100)
.map(plusOne)
.map(plusOne) // { value: 102 }
这个是时候再发挥我们从小被培养的找规律能力，我们发现：
两者的结构类似，拥有一个方法对内部的数据进行操作
两者均可进行链式调用
通过以上两点可以得到一个结论，之所以引入函子，可以解决链式调用的问题，但是光有一个函子不够呀，函子只能实现同步的链式调用，这时候另外一个原材料观察者模式就出场了。
观察者模式
先看一个简单的观察者模式实现：
class Observer {
constructor() {
this.callbacks = []
this.notify = value => {
this.callbacks.forEach(observe => observe(value))
}
}
subscribe = observer => {
this.callbacks.push(observer)
}
}
这时候聪明的人一下就发现了，这个 notify 和 subscribe 不就是 resolve 和 then 嘛！
俺のターン！ドロー！魔法発动！
c21f67221889af25edb36826285ce8aa_hd.jpg
ターンエンド！
class Promise {
constructor(executor) {
this.value = undefined
this.callbacks = []
// 相当于 notify
const resolve = value => {
this.value = value
this.callbacks.forEach(callback => callback())
}
executor(resolve)
}
// 相当于 subscribe 或 map
then = onFulfilled => new Promise(resolve => {
this.callbacks.push(() => resolve(onFulfilled(this.value)))
})
}
融合后的初级 Promise 已经具有异步链式调用的能力了比如：
const promise = new Promise(resolve => {
setTimeout(() => {
resolve(100)
}, 500)
})
.map(plusOne)
.map(plusOne)
// { value: 102 }
但是当我们进行一些骚操作时，依然会出问题：
const promise = new Promise(resolve => {
setTimeout(() => {
resolve(100)
resolve(1000)
}, 500)
})
.map(plusOne)
.map(plusOne)
// { value: 1002 }
为了解决这个问题，我们还需要一个原材料状态。
篇幅有限，这一部分更细致的转换过程，我的 repo 都有记录。
三、加入原材料 状态
众所周知，“青眼究极龙需要三条青眼白龙”，为了解决上一部分留下的问题，这一部分，需要给 Promise 加入状态这个原材料。
class Promise {
static PENDING = 'PENDING'
static FULFILLED = 'FULFILLED'
constructor(executor) {
this.value = undefined
this.callbacks = []
this.status = Promise.PENDING
// 一系列操作（状态的改变，成功回调的执行）
const resolve = value => {
// 只有处于 pending 状态的 promise 能调用 resolve
if (this.status === Promise.PENDING) {
// resolve 调用后，status 转为 fulfilled
this.status = Promise.FULFILLED
// 储存 fulfilled 的终值
this.value = value
// 一旦 resolve 执行，调用储存在回调数组里的回调
this.callbacks.forEach(callback => callback())
}
}
executor(resolve)
}
then = onFulfilled =>
new Promise(resolve => {
// 当 status 为执行态（Fulfilled）时
if (this.status === Promise.FULFILLED) {
resolve(onFulfilled(this.value))
}
// 当 status 为 Pending 时
if (this.status === Promise.PENDING) {
// 将 onFulfilled 存入回调数组
this.callbacks.push(() => resolve(onFulfilled(this.value)))
}
})
}
至此，通过三大原材料构建出的 Promise 就完成了，当然，还有很多功能没有实现，鲁迅曾经说过：“要站在巨人的肩膀上看问题。”，下一步，就需要 Promise/A+ 规范来来帮助我们实现一个具有完整功能的 Promise。
四、打开设计图纸 Promise/A+ 规范
剑来！ Promise/A+ 规范，接下来的操作，需要跟着它一步一步进行。
1、加入拒绝态以及处理（reject and onRejected）
其实这一步不用规范我们也知道，Promise 拥有的终态有 fulfilled 和 rejected 两种，所以要把剩下的 rejected 以及一些相关操作给补上。
class Promise {
......
static REJECTED = 'REJECTED'
constructor(executor) {
this.value = undefined
this.reason = undefined
this.onFulfilledCallbacks = []
this.onRejectedCallbacks = []
this.status = PromiseFunctorWithTwoStatus.PENDING
// 成功后的一系列操作（状态的改变，成功回调的执行）
const resolve = value => {
......
}
// 失败后的一系列操作（状态的改变，失败回调的执行）
const reject = reason => {
// 只有处于 pending 状态的 promise 能调用 resolve
if (this.status === Promise.PENDING) {
// reject 调用后，status 转为 rejected
this.status = Promise.REJECTED
// 储存 rejected 的拒因
this.reason = reason
// 一旦 reject 执行，调用储存在失败回调数组里的回调
this.onRejectedCallbacks.forEach(onRejected => onRejected())
}
}
executor(resolve, reject)
}
then = (onFulfilled, onRejected) =>
new Promise(resolve => {
// 当 status 为执行态（Fulfilled）时
......
// 当 status 为拒绝态（Rejected）时
if (this.status === PromiseFunctorWithTwoStatus.REJECTED) {
reject(onRejected(this.reason))
}
// 当 status 为 Pending 时
if (this.status === Promise.PENDING) {
// 将 onFulfilled 存入回调数组
this.onFulfilledCallbacks.push(() => resolve(onFulfilled(this.value)))
// 将 onRejected 存入失败回调数组
this.onRejectedCallbacks.push(() => reject(onRejected(this.reason)))
}
})
}
2、加入核心 resolvePromise 方法实现解决过程
Promise 解决过程是一个抽象的操作，其需输入一个 promise 和一个值，我们表示为 [[Resolve]](promise, x)，如果 x 有 then 方法且看上去像一个 Promise ，解决程序即尝试使 promise 接受 x 的状态；否则其用 x 的值来执行 promise 。
这种 thenable 的特性使得 Promise 的实现更具有通用性：只要其暴露出一个遵循 Promise/A+ 协议的 then 方法即可；这同时也使遵循 Promise/A+ 规范的实现可以与那些不太规范但可用的实现能良好共存。
根据规范的描述，我们依照他给的实现步骤，写出代码实现：
class Promise {
......
static resolvePromise = (anotherPromise, x, resolve, reject) => {
// 如果 onFulfilled 或者 onRejected 返回一个值 x ，则运行下面的 Promise 解决过程：[[Resolve]](promise2, x)
// 运行 [[Resolve]](promise, x) 需遵循以下步骤：
// 如果 promise 和 x 指向同一对象，以 TypeError 为拒因拒绝执行 promise 以防止循环引用
if (anotherPromise === x) {
return reject(new TypeError('Chaining cycle detected for promise'))
}
// 如果 x 为 Promise ，则使 promise 接受 x 的状态
if (x instanceof Promise) {
x.then(
// 如果 x 处于执行态，用相同的值执行 promise
value => {
return Promise.resolvePromise(anotherPromise, value, resolve, reject)
},
// 如果 x 处于拒绝态，用相同的拒因拒绝 promise
reason => {
return reject(reason)
}
)
// 如果 x 为对象或者函数
} else if (x !== null && (typeof x === 'object' || typeof x === 'function')) {
let called = false
try {
// 把 x.then 赋值给 then（这步我们先是存储了一个指向 x.then 的引用，然后测试并调用该引用，以避免多次访问 x.then 属性。这种预防措施确保了该属性的一致性，因为其值可能在检索调用时被改变。）
const then = x.then
// 如果 then 是函数，将 x 作为函数的作用域 this 调用之。传递两个回调函数作为参数，
if (typeof then === 'function') {
then.call(
x,
// 第一个参数叫做 resolvePromise ，
value => {
// 如果 resolvePromise 和 rejectPromise 均被调用，或者被同一参数调用了多次，则优先采用首次调用并忽略剩下的调用
if (called) {
return
}
called = true
// 如果 resolvePromise 以值 y 为参数被调用，则运行 [[Resolve]](promise, y)
return Promise.resolvePromise(
anotherPromise,
value,
resolve,
reject
)
},
// 第二个参数叫做 rejectPromise
reason => {
// 如果 resolvePromise 和 rejectPromise 均被调用，或者被同一参数调用了多次，则优先采用首次调用并忽略剩下的调用
if (called) {
return
}
called = true
// 如果 rejectPromise 以拒因 r 为参数被调用，则以拒因 r 拒绝 promise
return reject(reason)
}
)
} else {
//如果 then 不是函数，以 x 为参数执行 promise
return resolve(x)
}
} catch (error) {
// 如果调用 then 方法抛出了异常 e, 如果 resolvePromise 或 rejectPromise 已经被调用，则忽略之
if (called) {
return
}
called = true
// 如果取 x.then 的值时抛出错误 e ，则以 e 为拒因拒绝 promise
return reject(error)
}
} else {
// 如果 x 不为对象或者函数，以 x 为参数执行 promise
return resolve(x)
}
}
......
}
同时，我们要对之前的 Promise 里的 resolve 方法进行改造：
class Promise {
......
constructor(executor) {
......
// 成功后的一系列操作（状态的改变，成功回调的执行）
const resolve = x => {
const __resolve = value => {
// 只有处于 pending 状态的 promise 能调用 resolve
if (this.status === Promise.PENDING) {
// resolve 调用后，status 转为 fulfilled
this.status = Promise.FULFILLED
// 储存 fulfilled 的终值
this.value = value
// 一旦 resolve 执行，调用储存在成功回调数组里的回调
this.onFulfilledCallbacks.forEach(onFulfilled => onFulfilled())
}
}
return Promise.resolvePromise.call(this, this, x, __resolve, reject)
}
......
}
......
}
class Promise {
......
then = (onFulfilled, onRejected) => {
// then 方法必须返回一个 promise 对象
const anotherPromise = new Promise((resolve, reject) => {
// 封装处理链式调用的方法
const handle = (fn, argv) => {
// 确保 onFulfilled 和 onRejected 方法异步执行
setTimeout(() => {
try {
const x = fn(argv)
return Promise.resolvePromise(anotherPromise, x, resolve, reject)
} catch (error) {
return reject(error)
}
})
}
// 当 status 为执行态（Fulfilled）时
if (this.status === Promise.FULFILLED) {
// 则执行 onFulfilled，value 作为第一个参数
handle(onFulfilled, this.value)
}
// 当 status 为拒绝态（Rejected）时
if (this.status === Promise.REJECTED) {
// 则执行 onRejected，reason 作为第一个参数
handle(onRejected, this.reason)
}
// 当 status 为 Pending 时
if (this.status === Promise.PENDING) {
// 将 onFulfilled 存入成功回调数组
this.onFulfilledCallbacks.push(() => {
handle(onFulfilled, this.value)
})
// 将 onRejected 存入失败回调数组
this.onRejectedCallbacks.push(() => {
handle(onRejected, this.reason)
})
}
})
return anotherPromise
}
......
}
3、加入 其它方法 完善周边
Promise 的主体已经写好了，接下来要实现其他的一些辅助方法来完善它。
catch
catch = onRejected => {
return this.then(null, onRejected)
}
finally
finally = fn => {
return this.then(
value => {
setTimeout(fn)
return value
},
reason => {
setTimeout(fn)
throw reason
}
)
}
resolve
static resolve = value => new Promise((resolve, reject) => resolve(value))
reject
static reject = reason => new Promise((resolve, reject) => reject(reason))
all
static all = promises => {
if (!isArrayLikeObject(promises)) {
throw new TypeError(
`${
typeof promises === 'undefined' ? '' : typeof promises
} ${promises} is not iterable (cannot read property Symbol(Symbol.iterator))`
)
}
// 实现的 promise 基于 macroTask 的 setTimeout 实现，需要 async/await 调节执行顺序
// 原生的 promise 基于 microTask 实现，执行顺序是正确的，不需要 async/await
return new Promise(async (resolve, reject) => {
const result = []
for (const promise of promises) {
await Promise.resolve(promise).then(resolvePromise, rejectPromise)
}
return resolve(result)
function resolvePromise(value) {
if (value instanceof Promise) {
value.then(resolvePromise, rejectPromise)
} else {
result.push(value)
}
}
function rejectPromise(reason) {
return reject(reason)
}
})
}
race
static race = promises => {
if (!isArrayLikeObject(promises)) {
throw new TypeError(
`${
typeof promises === 'undefined' ? '' : typeof promises
} ${promises} is not iterable (cannot read property Symbol(Symbol.iterator))`
)
}
return new Promise((resolve, reject) => {
for (const promise of promises) {
Promise.resolve(promise).then(
value => resolve(value),
reason => reject(reason)
)
}
})
}
4、加入一些健壮性代码
这一部分基本上属于修修补补了，加强 Promise 的健壮性
校验 executor
constructor(executor) {
// 参数校验
if (typeof executor !== 'function') {
throw new TypeError(`Promise resolver ${executor} is not a function`)
}
}
利用 Maybe函子 的思想，校验 onFulfilled 和 onRejected
then = (onFulfilled, onRejected) => {
// 如果 onFulfilled 不是函数，其必须被“忽略”
onFulfilled =
typeof onFulfilled === 'function' ? onFulfilled : value => value
// 如果 onFulfilled 不是函数，其必须被“忽略”
onRejected =
typeof onRejected === 'function'
? onRejected
: error => {
throw error
}
}
五、装修为 Typescript 风格
这一部分就不写上来了，repo 里有记录。
六、测试是否符合 Promise/A+ 规范
我们通过一个库来检测写好的 Promise：
添加需要的胶水代码：
class Promise {
......
static defer = () => {
let dfd: any = {}
dfd.promise = new Promise((resolve, reject) => {
dfd.resolve = resolve
dfd.reject = reject
})
return dfd
}
static deferred = Promise.defer
......
}
npm i promises-aplus-tests -D
npx promises-aplus-tests promise.js
QQ20191215-201043.png
总结
最近在翻阅资料的过程中，真实地感悟到什么是“温故而知新”和“前端的知识虽然杂但是都有联系”。本来 Promise 的实现都被写烂了，但是在学习函数式编程的时候居然又绕回来了，这种感觉实在奇妙，让人不禁佩服第一个产生 Promise 想法的人。Promise 将 函子(functor) 和 观察者模式
相结合，加以 状态 、Promise 的解决过程 进行改造，最终得以实现一个异步解决方案。
篇幅有限，难免一些错误，欢迎探讨和指教~
附一个 GitHub 完整的 repo 地址：https://github.com/LazyDuke/ts-promise-from-functor-observer
后记
这是一个系列，系列文章：
手撕源码系列 —— lodash 的 debounce 与 throttle
手撕源码系列 —— 函子 + 观察者模式 + 状态 = Promise
手撕源码系列 —— 浅拷贝和深拷贝的完全实现（未完成）
手撕源码系列 —— 老生常谈的call、apply、bind和new（未完成）
发表评论 (审核通过后显示评论)：