缓解异步编程的不适

在最初学习 MFC 编程时，回调函数便是遇到的第一个难点。看着书中的定义 —— “ callback 函数虽然由你来设计，但是永远不会也不该被你调用，它们是为 Windows 系统准备的。” 我一脸的蒙圈。但是通过长时间的磨（wu）炼（jie），我终于在记忆中深深的烙上了不可缓解的不适，可谁曾想到这种不适延续到了 JS 。

回想 MFC 的基本思想，即 以消息为基础，以事件驱动之 。Windows 将事件转化为消息传递给 消息循环 ，再由 消息循环 分发给 窗口函数 来处理。 窗口函数 便是回调函数，因为你不知道用户何时会产生事件，而 Windows 会清楚。 如一个 Button 的点击处理逻辑由你来设计，而调用则交给了 Windows 。 来看张图，我知道它如此熟悉，但绝不是 NodeJS：

理解

回到 JS 细数回调函数的几种运用：

• 回调函数处理同步

const arr = [10, 20, 30, 40, 50, 60]

console.log("before")
let callback = arg => { // 回调逻辑
    console.log(arg)
}
arr.forEach(callback) // forEach 调用回调逻辑
console.log("after")
复制代码

• 当然也可以处理异步

console.log("before")
let callback = () => {  // 回调逻辑
    console.log('hello from callback');
}
setTimeout(callback, 1000) // setTimeout 调用回调逻辑
console.log("after")
复制代码

• NodeJS 中，添加了订阅过程：

const readable = getReadableStreamSomehow() //  业务逻辑如 fs.createReadStream('foo.txt')
function nextDataCallback (chunk) {  // 回调逻辑
  console.log(`Received ${chunk.length} bytes of data`)
}

function errorCallback (err) {
  console.error(`Bad stuff happened: ${err}.`)
}

function doneCallback () {
  console.log(`There will be no more data.`)
}

readable.on('data', nextDataCallback)  // readable  调用回调逻辑
readable.on('error', errorCallback)
readable.on('end', doneCallback)
复制代码

相信大家不陌生 NodeJS 的 Event Loop 机制，它和 MFC 的处理非常相似：

因为在 NodeJS 中 Event Loop 不同于 MFC 中的 消息循环 无须自己编写，所以需要在回调函数和 Event Loop 中建立联系，这就是添加了订阅过程的原因。

通过上述几种运用可以看出，虽然回调函数在 JS 中不再由 Windows 调用，但它依然遵循了 “你定义，但不是你调用” 的原则。同时在处理异步时，所被人诟病的回调地狱也促使新的处理方式的诞生。

Promise

Promise 构建在回调函数的基础上，实现了先执行异步再传递回调的方式。而这里只想简单说明回调函数和 Promise 的联系，不去纠结实现的细节，如需深入请参考其它文献。

在异步应用中，Promise 将分支假设为只有两种结果：成功或失败，我们先为此设计两个回调函数来响应结果：

function fulfilledCallback (data) {
  console.log(data)
}

function errorCallback (error) {
  console.log(error)
}
复制代码

添加业务逻辑：

const executor = (resolve, reject) => {
  setTimeout(function(){
    resolve('success')
  },1000)
}
复制代码

尝试执行：

executor(fulfilledCallback, errorCallback)
// wait one second
// 'success'
复制代码

现在我们将业务执行封装到函数中：

function someFun(executor) {
  executor(fulfilledCallback, errorCallback)
}

someFun(executor)
// wait one second
// 'success'
复制代码

在函数内添加状态表示业务执行结果：

function someFun(executor) {
  this.status = 'pending' // 初始状态
  this.value = undefined // 成功执行的数据
  this.reason = undefined // 失败执行的原因
  executor(fulfilledCallback, errorCallback)
}
复制代码

建立状态和回调函数的联动：

function someFun(executor) {
  this.status = 'pending' // 初始状态
  this.value = undefined // 成功执行的数据
  this.reason = undefined // 失败执行的原因
  function resolve(value) {
    self.status = 'resolved'
    self.value = value
    fulfilledCallback(self.value)
  }

  function reject(reason) {
    self.status = 'rejected'
    self.reason = reason
    errorCallback(self.reason)
  }
  executor(resolve, reject)
}
复制代码

添加方法 then 实现回调函数一般化，同时给函数改个名：

function myPromise(executor) {
  let self = this
  self.status = 'pending'
  self.value = undefined
  self.reason = undefined
  self.fulfilledCallbacks = []
  self.errorCallbacks = []

  self.then = function(fulfilledCallback, errorCallback) {
    self.fulfilledCallbacks.push(fulfilledCallback)
    self.errorCallbacks.push(errorCallback)
  }
  
  function resolve(value) {
    self.status = 'resolved'
    self.value = value
    self.fulfilledCallbacks.forEach(fn => {
      fn(self.value)
    })
  }

  function reject(reason) {
    self.status = 'rejected'
    self.reason = reason
    self.errorCallbacks.forEach(fn => {
      fn(self.reason)
    })
  }

  executor(resolve, reject)
}
复制代码

再次尝试执行：

function otherFulfilledCallback (data) {
  console.log('other' + ' ' + data)
}

function otherErrorCallback (error) {
  console.log('other' + ' ' +  error)
}

const p = new myPromise(executor)
p.then(otherFulfilledCallback, otherErrorCallback)

// wait one second
// 'other success'
复制代码

再次重申，这里只想研究回调函数如何封装进 Promise 的原理。

RxJS

现在我们将 NodeJS 的回调函数进行改进，首先放弃显式的订阅过程，将回调函数传递给业务调用：

function nextCallback(data) {
  // ...
}

function errorCallback(data) {
  // ...
}

function completeCallback(data) {
  // ...
}

giveMeSomeData(
  nextCallback,
  errorCallback,
  completeCallback
)
复制代码

补全回调回调函数，添加一个可以执行的业务：

function nextCallback(data) {
  console.log(data)
}

function errorCallback(err) {
  console.error(err)
}

function completeCallback(data) {
  console.log('done')
}

function giveMeSomeData(nextCB, errorCB, completeCB) {
  [10, 20, 30].forEach(nextCB)
}

giveMeSomeData(
  nextCallback,
  errorCallback,
  completeCallback
)
复制代码

将回调函数封装进对象中，同时将业务名称改为 subscribe ，也封装进 observable 对象：

const observable = {
  subscribe: function subscribe(ob) {
    [10, 20, 30].forEach(ob.next)
    ob.complete()
  }
}

const observer = {
  next: function nextCallback(data) {
    console.log(data)
  },
  error: function errorCallback(err) {
    console.error(err)
  },
  complete: function completeCallback(data) {
    console.log('done')
  }
}

observable.subscribe(observer)
复制代码

执行代码看看结果：

10
20
30
done
复制代码

再再次重申，这里只想研究回调函数如何封装进 Observable 的原理。