项目代码：github.com/Haixiang6123/my-promise-poller
轮询，一个前端非常常见的操作，然而对于很多人来说第一反应竟然还是用 setInterval 来实现， setInterval 作为轮询是不稳定的。下面就带大家一起写一个 promise-poller 的轮子吧。
从零开始先从上面说的 setInterval 的方法开始写起，一个最 Low 的轮询如下：
const promisePoller = (taskFn: Function, interval: number) => {
setInterval(() => {
taskFn();
}, interval)
}
第一个参数为轮询任务，第二参数为时间间隔，so easy。
刚刚也说了，setInterval 是不稳定的。用 setTimeout 迭代调用来做轮询会更稳定，面试题常见操作，so easy。
interface Options {
taskFn: Function
interval: number
}
const promisePoller = (options: Options) => {
const {taskFn, interval} = options
const poll = () => {
setTimeout(() => {
taskFn()
poll()
}, interval)
}
poll()
}
上面还把入参封成 Options 类型，更容易扩展入参类型。
这样的代码我们还是不满意，受不了 setTimeout 里又一个回调，太丑了。因此，可以把 setTimeout 封装成一个 delay 函数，delay 完成再去调用 poll 就好了。
export const delay = (interval: number) => new Promise(resolve => {
setTimeout(resolve, interval)
})
const promisePoller = (options: Options) => {
const {taskFn, interval} = options
const poll = () => {
taskFn()
delay(interval).then(poll) // 使用 delay 替换 setTimeout 的回调
}
poll()
}
是不是变干净多了？
promisify即然这个轮子的名字都带有 “promise”，那 promisePoller 函数肯定要返回一个 Promise 呀。这一步就要把这个函数 promisify。
首先返回一个 Promise。
const promisePoller = (options: Options) => {
const {taskFn, interval, masterTimeout} = options
return new Promise((resolve, reject) => { // 返回一个 Promise
const poll = () => {
const result = taskFn()
delay(interval).then(poll)
}
poll()
})
}
那问题来了：什么时候该 reject ？什么时候该 resolve 呢？自然是整个轮询失败就 reject，整个轮询成功就 resolve 呗。
先看 reject 时机：整个轮询失败一般是 timeout 了就凉了呗，所以这里加个 masterTimeout 到 Options 中，表示整体轮询的超时时间，再为整个轮询过程加个 setTimeout 计时器。
interface Options {
taskFn: Function
interval: number
masterTimeout?: number // 整个轮询过程的 timeout 时长
}
const promisePoller = (options: Options) => {
const {taskFn, interval, masterTimeout} = options
let timeoutId
return new Promise((resolve, reject) => {
if (masterTimeout) {
timeoutId = setTimeout(() => {
reject('Master timeout') // 整个轮询超时了
}, masterTimeout)
}
const poll = () => {
taskFn()
delay(interval).then(poll)
}
poll()
})
}
再看 resolve 时机：执行到最后一次轮询任务就说明整个轮询成功了嘛，那怎么才知道这是后一次的轮询任务呢？呃，我们并不能知道，只能通过调用方告诉我们才知道，所以加个 shouldContinue 的回调让调用方告诉我们当前是否应该继续轮询，如果不继续就是最后一次了嘛。
interface Options {
taskFn: Function
interval: number
shouldContinue: (err: string | null, result: any) => boolean // 当次轮询后是否需要继续
masterTimeout?: number
}
const promisePoller = (options: Options) => {
const {taskFn, interval, masterTimeout, shouldContinue} = options
let timeoutId: null | number
return new Promise((resolve, reject) => {
if (masterTimeout) {
timeoutId = window.setTimeout(() => {
reject('Master timeout') // 整个轮询过程超时了
}, masterTimeout)
}
const poll = () => {
const result = taskFn()
if (shouldContinue(null, result)) {
delay(interval).then(poll) // 继续轮询
} else {
if (timeoutId !== null) { // 不需要轮询，有 timeoutId 则清除
clearTimeout(timeoutId)
}
resolve(result) // 最后一个轮询任务了，结束并返回最后一次 taskFn 的结果
}
}
poll()
})
}
至此，一个 Promisify 后的 poller 函数已经大体完成了。还有没有得优化呢？有！
轮询任务的 timeout刚刚提到 masterTimeout，相对地，也应该有轮询单个任务的 timeout，所以，在 Options 里加个 taskTimeout 字段吧。
不对，等等！上面好像我们默认 taskFn 是同步的函数呀，timeout 一般针对异步函数设计的，这也提示了我们 taskFn 应该也要支持异步函数才行。所以，在调用 taskFn 的时候，要将其结果 promisify，然后对这个 promise 进行 timeout 的检测。
interface Options {
taskFn: Function
interval: number
shouldContinue: (err: string | null, result: any) => boolean
masterTimeout?: number
taskTimeout?: number // 轮询任务的 timeout
}
// 判断该 promise 是否超时了
const timeout = (promise: Promise, interval: number) => {
return new Promise((resolve, reject) => {
const timeoutId = setTimeout(() => reject('Task timeout'), interval)
promise.then(result => {
clearTimeout(timeoutId)
resolve(result)
})
})
}
const promisePoller = (options: Options) => {
const {taskFn, interval, masterTimeout, taskTimeout, shouldContinue} = options
let timeoutId: null | number
return new Promise((resolve, reject) => {
if (masterTimeout) {
timeoutId = window.setTimeout(() => {
reject('Master timeout')
}, masterTimeout)
}
const poll = () => {
let taskPromise = Promise.resolve(taskFn()) // 将结果 promisify
if (taskTimeout) {
taskPromise = timeout(taskPromise, taskTimeout) // 检查该轮询任务是否超时了
}
taskPromise
.then(result => {
if (shouldContinue(null, result)) {
delay(interval).then(poll)
} else {
if (timeoutId !== null) {
clearTimeout(timeoutId)
}
resolve(result)
}
})
.catch(error => {
})
}
poll()
})
}
上面一共完成了三步：
- 将 taskFn 的结果 promisify
- 添加 timeout 函数用于判断 taskFn 是否超时（对于同步函数其实一般来说不会 timeout，因为结果是马上返回的）
- 判断 taskFn 是否超时，超时了直接 reject，会走到 taskPromise 的 catch 里
那如果真的超时了，timeout reject 了之后干啥呢？当然是告诉主流程的轮询说：哎，这个任务超时了，我要不要重试一下啊。因此，这里又要引入一个重试的功能了。
重试首先，在 Options 加个 retries 的字段表示可重试的次数。
interface Options {
taskFn: Function
interval: number
shouldContinue: (err: string | null, result?: any) => boolean
masterTimeout?: number
taskTimeout?: number
retries?: number // 轮询任务失败后重试次数
}
接着在 catch 里，判断 retries 是否为 0（重试次数还没用完） 和 shouldContinue 的值是否为 true（我真的要重试啊），以此来确定是否真的需要重试。只有两者都为 true 时才重试。
const promisePoller = (options: Options) => {
...
let rejections: Array = []
let retriesRemain = retries
return new Promise((resolve, reject) => {
...
const poll = () => {
...
taskPromise
.then(result => {
...
})
.catch(error => {
rejections.push(error) // 加入 rejections 错误列表
if (--retriesRemain === 0 || !shouldContinue(error)) { // 判断是否需要重试
reject(rejections) // 不重试，直接失败
} else {
delay(interval).then(poll); // 重试
}
})
}
poll()
})
}
上面还添加 rejections 变量用于存放多个 error 信息。这样的设计是因为有可能 10 个任务里 2 个失败了，那最后就要把 2 个失败的信息都返回，因此需要一个数组存放错误信息。
这里还一个优化点：我们常常看到别人页面获取数据失败都会显示 1/3 获取… 2/3 获取… 3/3 获取… 直到真的获取失败，相当于有个进度条，这样对用户也比较好。所以，在 Options 可以提供一个回调，每次 retriesRemain 要减少时调用一下就好了。
interface Options {
taskFn: Function
interval: number
shouldContinue: (err: string | null, result?: any) => boolean
progressCallback?: (retriesRemain: number, error: Error) => unknown // 剩余次数回调
masterTimeout?: number
taskTimeout?: number
retries?: number // 轮询任务失败后重试次数
}
const promisePoller = (options: Options) => {
...
let rejections: Array = []
let retriesRemain = retries
return new Promise((resolve, reject) => {
...
const poll = () => {
...
taskPromise
.then(result => {
...
})
.catch(error => {
rejections.push(error) // 加入 rejections 错误列表
if (progressCallback) {
progressCallback(retriesRemain, error) // 回调获取 retriesRemain
}
if (--retriesRemain === 0 || !shouldContinue(error)) { // 判断是否需要重试
reject(rejections) // 不重试，直接失败
} else {
delay(interval).then(poll); // 重试
}
})
}
poll()
})
}
虽然 shouldContinue 已经可以有效地控制流程是否要中止，但是每次都要等下一次轮询开始之后才会判断，这样未免有点被动。如果可以在 taskFn 执行的时候就主动停止，那 promisePoller 就更灵活了。
而 taskFn 有可能是同步函数或者异步函数，对于同步函数，我们规定 return false 就停止轮询，对于异步函数，规定 reject("CANCEL_TOKEN") 就停止轮询。函数改写如下：
const CANCEL_TOKEN = 'CANCEL_TOKEN'
const promisePoller = (options: Options) => {
const {taskFn, masterTimeout, taskTimeout, progressCallback, shouldContinue, retries = 5} = mergedOptions
let polling = true
let timeoutId: null | number
let rejections: Array = []
let retriesRemain = retries
return new Promise((resolve, reject) => {
if (masterTimeout) {
timeoutId = window.setTimeout(() => {
reject('Master timeout')
polling = false
}, masterTimeout)
}
const poll = () => {
let taskResult = taskFn()
if (taskResult === false) { // 结束同步任务
taskResult = Promise.reject(taskResult)
reject(rejections)
polling = false
}
let taskPromise = Promise.resolve(taskResult)
if (taskTimeout) {
taskPromise = timeout(taskPromise, taskTimeout)
}
taskPromise
.then(result => {
...
})
.catch(error => {
if (error === CANCEL_TOKEN) { // 结束异步任务
reject(rejections)
polling = false
}
rejections.push(error)
if (progressCallback) {
progressCallback(retriesRemain, error)
}
if (--retriesRemain === 0 || !shouldContinue(error)) {
reject(rejections)
} else if (polling) { // 再次重试时，需要检查 polling 是否为 true
delay(interval).then(poll);
}
})
}
poll()
})
}
上面代码判断了 taskFn 的返回值是否为 false，在 catch 里判断 error 是否为 CANCEL_TOKEN。如果 taskFn 主动要求中止轮询，那么设置 polling 为 false 并 reject 整个流程。
这里还有个细节是：为了提高安全性，在重试的那里要再检查一次 polling 是否为 true 才重新 poll。
轮询策略目前我们设计的都是线性轮询的，一个 interval 搞定。为了提高扩展性，我们再提供另外 2 种轮询策略：linear-backoff 和 exponential-backoff，分别对 interval 的线性递增和指数递增，而非匀速不变。
先定好策略的一些默认参数：
export const strategies = {
'fixed-interval': {
defaults: {
interval: 1000
},
getNextInterval: function(count: number, options: Options) {
return options.interval;
}
},
'linear-backoff': {
defaults: {
start: 1000,
increment: 1000
},
getNextInterval: function(count: number, options: Options) {
return options.start + options.increment * count;
}
},
'exponential-backoff': {
defaults: {
min: 1000,
max: 30000
},
getNextInterval: function(count: number, options: Options) {
return Math.min(options.max, Math.round(Math.random() * (Math.pow(2, count) * 1000 - options.min) + options.min));
}
}
};
每种策略都有自己的参数和 getNextInterval 的方法，前者为起始参数，后者在轮询的时候实时获取下一次轮询的时间间隔。因为有了起始参数，Options 的参数也要改动一下。
type StrategyName = 'fixed-interval' | 'linear-backoff' | 'exponential-backoff'
interface Options {
taskFn: Function
shouldContinue: (err: Error | null, result?: any) => boolean // 当次轮询后是否需要继续
progressCallback?: (retriesRemain: number, error: Error) => unknown // 剩余次数回调
strategy?: StrategyName // 轮询策略
masterTimeout?: number
taskTimeout?: number
retries?: number
// fixed-interval 策略
interval?: number
// linear-backoff 策略
start?: number
increment?: number
// exponential-backoff 策略
min?: number
max?: number
}
在 poll 函数里就简单了，只需要在 delay 之前获取一下 nextInterval，然后 delay(nextInterval) 即可。
const promisePoller = (options: Options) => {
const strategy = strategies[options.strategy] || strategies['fixed-interval'] // 获取当前的轮询策略，默认使用 fixed-interval
const mergedOptions = {...strategy.defaults, ...options} // 合并轮询策略的初始参数
const {taskFn, masterTimeout, taskTimeout, progressCallback, shouldContinue, retries = 5} = mergedOptions
let polling = true
let timeoutId: null | number
let rejections: Array = []
let retriesRemain = retries
return new Promise((resolve, reject) => {
if (masterTimeout) {
timeoutId = window.setTimeout(() => {
reject(new Error('Master timeout'))
polling = false
}, masterTimeout)
}
const poll = () => {
let taskResult = taskFn()
if (taskResult === false) {
taskResult = Promise.reject(taskResult)
reject(rejections)
polling = false
}
let taskPromise = Promise.resolve(taskResult)
if (taskTimeout) {
taskPromise = timeout(taskPromise, taskTimeout)
}
taskPromise
.then(result => {
if (shouldContinue(null, result)) {
const nextInterval = strategy.getNextInterval(retriesRemain, mergedOptions) // 获取下次轮询的时间间隔
delay(nextInterval).then(poll)
} else {
if (timeoutId !== null) {
clearTimeout(timeoutId)
}
resolve(result)
}
})
.catch((error: Error) => {
if (error.message === CANCEL_TOKEN) {
reject(rejections)
polling = false
}
rejections.push(error)
if (progressCallback) {
progressCallback(retriesRemain, error)
}
if (--retriesRemain === 0 || !shouldContinue(error)) {
reject(rejections)
} else if (polling) {
const nextInterval = strategy.getNextInterval(retriesRemain, options) // 获取下次轮询的时间间隔
delay(nextInterval).then(poll);
}
})
}
poll()
})
}
这个 promisePoller 主要完成了：
- 基础的轮询操作
- 返回 promise
- 提供主动和被动中止轮询的方法
- 提供轮询任务重试的功能，并提供重试进度回调
- 提供多种轮询策略：fixed-interval, linear-backoff, exponential-backoff
以上就是 npm 包 [promise-poller] www.npmjs.com/package/promise-poller 的源码实现了。
