浅谈异步流程控制

异步流程可以说是前端工作中每天都会打交道的场景，绝大部分情况下的业务场景都不复杂，简单的Promise就可以应付。不过现在既然有这么多异步控制的武器了，假想一些场景练习一下熟练度也是很不错的，于是就有了这样一篇对异步控制的回顾文章。

首先简单模拟一个异步请求函数，接下来的所有场景的操作都以这个函数为基础，通常的异步流程组合也会由这种粒度的函数组成。

function fetch(api, data?) {
    return new Promise((resolve, reject) => {
        setTimeout(function() {
            resolve('newdata');
        }, 1000);
    })
}

继发

这可能是业务中碰到最多的场景了，简单来说就是异步操作B依赖于异步操作A的结果。那么通常情况下会这么写。

fetch(apiA, data).then((res) => {
    let newData = doSomething(res);
    return fetch(apiB, newData);
});

只有一两步操作还好，但当继发操作过长时，写下一连串Promise链显然不是一个好的实现，就可以考虑用循环处理串联部分,并将链各个节点的数据处理再封装为新的函数。

function fetchA(data) {
    return fetch(apiA, data).then((res) => doSomething(res));
}

function fetchB(data) {
    return fetch(apiB, data).then((res) => doSomething2(res));
}

// ...
let fetchers = [fetchA, fetchB, fetchC, ...];
let originData = {};
let promise = Promise.resolve(originData);
for (let i = 0; i < fetchers.length; i++) {
    promise = promise.then(fetchers[i]);
}

也可以使用更加简洁的reduce操作。

let fetchers = [fetchA, fetchB, fetchC, ...];
let originData = {};
fetchers.reduce((promise, fetch) => promise.then(fetch), Promise.resolve(originData));

与时俱进的，可以封装为async函数让可读性更好。

async function queueFetch(fetchers, originData) {
    let res = originData;
    for (let promise of fetchers) {
        res = await promise(res);
    }
    return await res;
}

并发

另外一种常见场景就是多个异步请求的结果相互不依赖，为了效率考虑就可以并发处理。最普遍的处理手段就是利用Promise.all。

Promise.all([fetch(apiA, dataA), fetch(apiB, dataB), ...]).then(res => {
    // ...
})

但这样会造成的一个问题就是一但某个promise出错，整个promise都会reject掉。另外当我们对请求返回的结果处理有顺序要求时，单纯的Promise.all也无法实现。

所以同样的，我们可以考虑循环处理Promise，不同的是在串联Promise链之前就将请求并发，同时对链的每个节点增加错误处理，这样也保证了按数组顺序处理并发请求返回的结果。

let fetchers = [fetchA(apiA, dataA), fetchB(apiB, dataB), ...];
let promise = Promise.resolve(originData);
for (let i = 0; i < fetchers.length; i++) {
    promise = promise.then(() => fetchers[i]).catch(err => console.error(err));
}

同样的也可以转化为reduce和async写法，就不赘述了。

最大并发数限制

对于并发而言还需要考虑的一个场景是，并发数量肯定不能是无限的，首先有浏览器限制，同时还有性能方面的考虑，最常见的场景就是爬虫，大量并发请求下通常会设置最大并发数，当前部分请求完成后，小于最大并发数时再继续请求。

这种场景非常适合用Promise.race实现，维护一个最大并发队列，通过race方法找到最先完成的Promise推出，并将排队中的请求推入，直至不再有排队请求，再用all方法完成队列中所有请求。其中需要注意的一个关键点就是如何在并发队列中找到race出来的promise，可以选择一层高阶函数重构返回结果，增加一个标识实现。

function limitFetch(fetchers, limit) {
    const wrapper = function(fetcher) {
        const promise = fetcher().then(res => ({
            data: res,
            index: promise
        }));
        return promise;
    };

    let promiseQueue = fetchers.splice(0, limit).map(wrapper);

    return fetchers.reduce((promisePop, fetcher) => {
        return promisePop.then(() => Promise.race(promiseQueue))
            .catch(err => console.error(err))
            .then(res => {
            let pos = promiseQueue.findIndex(item => item === res.index);
            promiseQueue.splice(pos, 1);
            promiseQueue.push(wrapper(fetcher));
        });
    }, Promise.resolve()).then(() => Promise.all(promisesQueue));
}

另外的思路

以上应该能覆盖比较常见的异步流程控制场景，其实属于比较简单的异步处理，可以看见我们主要用了JS中的主流异步控制工具Promise和async，还是写了不少代码，虽然可以应付大部分业务，但总有一小部分异步场景会比上述情况复杂得多，这种时候就应该考虑专为异步而生的强大武器——RxJS了。具体使用以前也写过一些文章就不在这里介绍，不妨看看上述场景通过RxJS实现是如何。

继发

let fetchers = [fetchA, fetchB, fetchC, ...];
let originData = {};
const subject = new BehaviorSubject(Observable.from(fetchers.pop()(originData)));

subject.switch().subscribe(data => {
    let fetcher = fetchers.pop();
    if (fetcher) {
        subject.next(Observable.from(fetcher(data)));
    } else {
        // do something
    }
})

并发

let fetchers = [fetch(apiA, dataA), fetch(apiB, dataB), ...]
Observable.from(fetchers)
    .mergeMap(promise => Observable.from(promise))
        .do(data => {
           // do something for a fetcher
        })
        .reduce((acc, cur) => {
            acc.push(cur);
            return acc;
        }, [])
        .subscribe(dataArr => {
            // do something for all fetchers
        })

虽然看上去好像比Promise版本更复杂，主要是涉及到新机制和操作符的引入。当异步场景更复杂时，RxJS带来的这套机制就会越来越省代码。