異步隊(duì)列
很多面試的時(shí)候會問一個(gè)問題，就是怎么讓異步函數(shù)可以順序執(zhí)行。方法有很多，callback，promise，觀察者，generator，async/await，這些 JS 中處理異步編程的，都可以做到這種串行的需求。但是很麻煩的是，處理起來是挺麻煩的，你要不停的手動在上一個(gè)任務(wù)調(diào)用下一個(gè)任務(wù)。比如 promise，像這樣：
a.then(() => b.then(() => c.then(...)))復(fù)制代碼代碼嵌套的問題，有點(diǎn)嚴(yán)重。所以要是有一個(gè)隊(duì)列就好了，往隊(duì)列里添加異步任務(wù)，執(zhí)行的時(shí)候讓隊(duì)列開始 run 就好了。先制定一下 API，我們有一個(gè) queue，隊(duì)列都在內(nèi)部維護(hù)，通過 queue.add 添加異步任務(wù)，queue.run 執(zhí)行隊(duì)列，可以先想想。
參照之前 express 中間件的實(shí)現(xiàn)，給異步任務(wù) async-fun 傳入一個(gè) next 方法，只有調(diào)用 next，隊(duì)列才會繼續(xù)往下走。那這個(gè) next 就至關(guān)重要了，它會控制隊(duì)列往后移一位，執(zhí)行下一個(gè) async-fun。我們需要一個(gè)隊(duì)列，來保存 async-fun，也需要一個(gè)游標(biāo)，來控制順序。
以下是我的簡單實(shí)現(xiàn)：
const queue = () => {
const list = []; // 隊(duì)列
let index = 0; // 游標(biāo)
// next 方法
const next = () => {
if (index >= list.length - 1) return;
// 游標(biāo) + 1
const cur = list[++index];
cur(next);
}
// 添加任務(wù)
const add = (...fn) => {
list.push(...fn);
}
// 執(zhí)行
const run = (...args) => {
const cur = list[index];
typeof cur === 'function' && cur(next);
}
// 返回一個(gè)對象
return {
add,
run,
}
}
// 生成異步任務(wù)
const async = (x) => {
return (next) => {// 傳入 next 函數(shù)
setTimeout(() => {
console.log(x);
next(); // 異步任務(wù)完成調(diào)用
}, 1000);
}
}
const q = queue();
const funs = '123456'.split('').map(x => async(x));
q.add(...funs);
q.run();// 1, 2, 3, 4, 5, 6 隔一秒一個(gè)。復(fù)制代碼我這里沒去構(gòu)造一個(gè) class，而是通過閉包的特性去處理的。queue 方法返回一個(gè)包含 add，run 的對象，add 即為像隊(duì)列中添加異步方法，run 就是開始執(zhí)行。在 queue 內(nèi)部，我們定義了幾個(gè)變量，list 用來保存隊(duì)列，index 就是游標(biāo)，表示隊(duì)列現(xiàn)在走到哪個(gè)函數(shù)了，另外，最重要的是 next 方法，它是控制游標(biāo)向后移動的。
run 函數(shù)一旦執(zhí)行，隊(duì)列即開始 run。一開始執(zhí)行隊(duì)列里的第一個(gè) async 函數(shù)，我們把 next 函數(shù)傳給了它，然后由 async 函數(shù)決定什么時(shí)候執(zhí)行 next，即開始執(zhí)行下一個(gè)任務(wù)。我們沒有并不知道異步任務(wù)什么時(shí)候才算完成，只能通過打成某種共識，來告知 queue 某個(gè)任務(wù)完成。就是傳給任務(wù)的 next 函數(shù)。其實(shí) async 返回的這個(gè)函數(shù)，有一個(gè)名字，叫 Thunk，后面我們會簡單介紹。
Thunk
thunk 其實(shí)是為了解決 “傳名調(diào)用” 的。就是我傳給函數(shù) A 一個(gè)表達(dá)式作參數(shù) x + 1，但是我不確定這個(gè) x + 1 什么時(shí)候會用到，以及會不會用到，如果在傳入就執(zhí)行，這個(gè)求值是沒有必要的。所以就出現(xiàn)了一個(gè)臨時(shí)函數(shù) Thunk，來保存這個(gè)表達(dá)式，傳入函數(shù) A 中，待需要時(shí)再調(diào)用。
const thunk = () => {
return x + 1;
};
const A = thunk => {
return thunk() * 2;
}復(fù)制代碼嗯... 其實(shí)就是一個(gè)回調(diào)函數(shù)...
暫停
其實(shí)只要某個(gè)任務(wù)，不繼續(xù)調(diào)用 next，隊(duì)列就已經(jīng)不會繼續(xù)往下走了。比如我們 async 任務(wù)里加一個(gè)判斷（通常是異步 io，請求的容錯(cuò)處理）：
// queue 函數(shù)不變，
// async 加限制條件
const async = (x) => {
return (next) => {
setTimeout(() => {
if(x > 3) {
console.log(x);
q.run(); //重試
return;
}
console.log(x);
next();
}, 1000);
}
}
const q = queue();
const funs = '123456'.split('').map(x => async(x));
q.add(...funs);
q.run();
//打印結(jié)果： 1, 2, 3, 4, 4，4， 4，4 一直是 4復(fù)制代碼當(dāng)執(zhí)行到第四個(gè)任務(wù)的時(shí)候，x 是 4 的時(shí)候，不再繼續(xù)，就可以直接 return，不再調(diào)用 next。也有可能是出現(xiàn)錯(cuò)誤，我們需要再重試，那就再調(diào)用 q.run 就可以了，因?yàn)橛螛?biāo)保存的就是當(dāng)前的 async 任務(wù)的索引。
另外，還有一種方式，就是添加 stop 方法。雖然感覺上面的方法就 OK 了，但是 stop 的好處在于，你可以主動的停止隊(duì)列，而不是在 async 任務(wù)里加限制條件。當(dāng)然，有暫停就有繼續(xù)了，兩種方式，一個(gè)是 retry，就是重新執(zhí)行上一次暫停的那個(gè)；另一個(gè)就是 goOn，不管上次最后一個(gè)如何，繼續(xù)下一個(gè)。上代碼：
const queue = () => {
const list = [];
let index = 0;
let isStop = false;
const next = () => {
// 加限制
if (index >= list.length - 1 || isStop) return;
const cur = list[++index];
cur(next);
}
const add = (...fn) => {
list.push(...fn);
}
const run = (...args) => {
const cur = list[index];
typeof cur === 'function' && cur(next);
}
const stop = () => {
isStop = true;
}
const retry = () => {
isStop = false;
run();
}
const goOn = () => {
isStop = false;
next();
}
return {
add,
run,
stop,
retry,
goOn,
}
}
const async = (x) => {
return (next) => {
setTimeout(() => {
console.log(x);
next();
}, 1000);
}
}
const q = queue();
const funs = '123456'.split('').map(x => async(x));
q.add(...funs);
q.run();
setTimeout(() => {
q.stop();
}, 3000)
setTimeout(() => {
q.goOn();
}, 5000)復(fù)制代碼其實(shí)還是加攔截... 只不過從 async 函數(shù)中，換到了 next 函數(shù)里面，利用 isStop 這個(gè)變量切換 true/false，開關(guān)暫停。我加了兩個(gè)定時(shí)器，一個(gè)是 3 秒后暫停，一個(gè)是 5 秒后繼續(xù)，（請忽略定時(shí)器的誤差），按道理應(yīng)該是隊(duì)列到三秒的時(shí)候，也就是第三個(gè)任務(wù)執(zhí)行完暫停，然后再隔 2 秒，繼續(xù)。結(jié)果打印到 3 的時(shí)候，停住，兩秒之后繼續(xù) 4，5，6.
兩種思路，請結(jié)合場景思考問題。
并發(fā)
上面的都是在做串行，假如 run 的時(shí)候我要并行呢... 也很簡單，把隊(duì)列一次性跑完就可以了。
// 為了代碼短一些，把 retry，goOn 先去掉了。
const queue = () => {
const list = [];
let index = 0;
let isStop = false;
let isParallel = false;
const next = () => {
if (index >= list.length - 1 || isStop || isParallel) return;
const cur = list[++index];
cur(next);
}
const add = (...fn) => {
list.push(...fn);
}
const run = (...args) => {
const cur = list[index];
typeof cur === 'function' && cur(next);
}
const parallelRun = () => {
isParallel = true;
for(const fn of list) {
fn(next);
}
}
const stop = () => {
isStop = true;
}
return {
add,
run,
stop,
parallelRun,
}
}
const async = (x) => {
return (next) => {
setTimeout(() => {
console.log(x);
next();
}, 1000);
}
}
const q = queue();
const funs = '123456'.split('').map(x => async(x));
q.add(...funs);
q.parallelRun();
// 一秒后全部輸出 1, 2, 3, 4, 5, 6復(fù)制代碼我添加了一個(gè) parallelRun 方法，用于并行，我覺得還是不要放到 run 函數(shù)里面了，抽象單元盡量細(xì)化還是。然后還加了一個(gè) isParallel 的變量，默認(rèn)是 false，考慮到 next 函數(shù)有可能會被調(diào)用，所以需要加一個(gè)攔截，保證不會處亂。
以上就是利用僅用 thunk 函數(shù)，結(jié)合 next 實(shí)現(xiàn)的異步隊(duì)列控制器，queue，跟你可以把 es6 代碼都改成 es5，保證兼容，當(dāng)然是足夠簡單的，不適用于負(fù)責(zé)的場景 ??，僅提供思路。
generator 與 co
為什么要介紹 generator，首先它也是用來解決異步回調(diào)的，另外它的使用方式也是調(diào)用 next 函數(shù)，generator 才會往下執(zhí)行，默認(rèn)是暫停狀態(tài)。yield 就相當(dāng)于上面的 q.add，往隊(duì)列中添加任務(wù)。所以我也打算一起介紹，來更好的拓寬思路。發(fā)散思維，相似的知識點(diǎn)做好歸納，然后某一天你就會突然有一種：原來是這么回事，原來 xxx 是借鑒子 yyy，然后你又去研究 yyy - -。
簡介 generator
簡單介紹回顧一下，因?yàn)橛型瑢W(xué)不經(jīng)常用，肯定會有遺忘。
// 一個(gè)簡單的栗子，介紹它的用法
function* gen(x) {
const y = yield x + 1;
console.log(y, 'here'); // 12
return y;
}
const g = gen(1);
const value = g.next().value; // {value: 2, done: false}
console.log(value); // 2
console.log(g.next(value + 10)); // {value: 12, done: true}復(fù)制代碼首先生成器其實(shí)就是一個(gè)通過函數(shù)體內(nèi)部定義迭代算法，然后返回一個(gè) iterator 對象。關(guān)于iterator，可以看我另一篇文章。
gen 執(zhí)行返回一個(gè)對象 g，而不是返回結(jié)果。g 跟其他 iterator 一樣，通過調(diào)用 next 方法，保證游標(biāo) + 1，并且返回一個(gè)對象，包含了 value（yield 語句的結(jié)果），和 done（迭代器是否完成）。另外，yield 語句的值，比如上面代碼中的 y，是下一次調(diào)用 next 傳入的參數(shù)，也就是 value + 10，所以是 12.這樣設(shè)計(jì)是有好處的，因?yàn)檫@樣你就可以在 generator 內(nèi)部，定義迭代算法的時(shí)候，拿到上次的結(jié)果（或者是處理后的結(jié)果）了。
但是 generator 有一個(gè)弊端就是不會自動執(zhí)行，TJ 大神寫了一個(gè) co，來自動執(zhí)行 generator，也就是自動調(diào)用 next。它要求 yield 后面的函數(shù)/語句，必須是 thunk 函數(shù)或者是 promise 對象，因?yàn)橹挥羞@樣才會串聯(lián)執(zhí)行完，這跟我們最開始實(shí)現(xiàn) queue 的思路是一樣的。co 的實(shí)現(xiàn)有兩種思想，一個(gè)是 thunk，一個(gè)是 promise，我們都來試一下。
Thunk 實(shí)現(xiàn)
還記得最開始的 queue 怎么實(shí)現(xiàn)的嗎，內(nèi)部定義 next 函數(shù)，來保證游標(biāo)的前進(jìn)，async 函數(shù)會接收 next，去執(zhí)行 next。到這里是一樣的，我們只要在 co 函數(shù)內(nèi)部定義一個(gè)同樣的 next 函數(shù)，來保證繼續(xù)執(zhí)行，那么 generator 是沒有提供索引的，不過它提供了 g.next 函數(shù)啊，所以我們只需要給 async 函數(shù)傳 g.next 不就好了，async 就是 yield 后面的語句啊，也就是 g.value。但是并不能直接傳 g.next，為什么？因?yàn)橄乱淮蔚?thunk 函數(shù)，要通過 g.next 的返回值 value 取到啊，木有 value，下一個(gè) thunk 函數(shù)不就沒了... 所以我們還是需要定義一個(gè) next 函數(shù)去包裝一下的。
上代碼：
const coThunk = function(gen, ...params) {
const g = gen(...params);
const next = (...args) => { // args 用于接收參數(shù)
const ret = g.next(...args); // args 傳給 g.next，即賦值給上一個(gè) yield 的值。
if(!ret.done) { // 去判斷是否完成
ret.value(next); // ret.value 就是下一個(gè) thunk 函數(shù)
}
}
next(); // 先調(diào)用一波
}
// 返回 thunk 函數(shù)的 asyncFn
const asyncFn = (x) => {
return (next) => { // 接收 next
const data = x + 1;
setTimeout(() => {
next && next(data);
}, 1000)
}
}
const gen = function* (x) {
const a = yield asyncFn(x);
console.log(a);
const b = yield asyncFn(a);
console.log(b);
const c = yield asyncFn(b);
console.log(c);
const d = yield asyncFn(c);
console.log(d);
console.log('done');
}
coThunk(gen, 1);
// 2, 3, 4, 5, done復(fù)制代碼這里定義的 gen，功能很簡單，就是傳入?yún)?shù) 1，然后每個(gè) asyncFn 異步累加，即多個(gè)異步操作串行，并且下一個(gè)依賴上一個(gè)的返回值。
promise 實(shí)現(xiàn)
其實(shí)思路都是一樣的，只不過調(diào)用 next，換到了 co 內(nèi)部。因?yàn)?yield 后面的語句是 promise 對象的話，我們可以在 co 內(nèi)部拿到了，然后在 g.next().value 的 then 語句執(zhí)行 next 就好了。
// 定義 co
const coPromise = function(gen) {
// 為了執(zhí)行后的結(jié)果可以繼續(xù) then
return new Promise((resolve, reject) => {
const g = gen();
const next = (data) => { // 用于傳遞，只是換個(gè)名字
const ret = g.next(data);
if(ret.done) { // done 后去執(zhí)行 resolve，即co().then(resolve)
resolve(data); // 最好把最后一次的結(jié)果給它
return;
}
ret.value.then((data) => { // then 中的第一個(gè)參數(shù)就是 promise 對象中的 resolve，data 用于接受并傳遞。
next(data); //調(diào)用下一次 next
})
}
next();
})
}
const asyncPromise = (x) => {
return new Promise((resolve) => {
setTimeout(() => {
resolve(x + 1);
}, 1000)
})
}
const genP = function* () {
const data1 = yield asyncPromise(1);
console.log(data1);
const data2 = yield asyncPromise(data1);
console.log(data2);
const data3 = yield asyncPromise(data2);
console.log(data3);
}
coPromise(genP).then((data) => {
setTimeout(() => {
console.log(data + 1); // 5
}, 1000)
});
// 一樣的 2, 3, 4, 5復(fù)制代碼其實(shí) co 的源碼就是通過這兩種思路實(shí)現(xiàn)的，只不過它做了更多的 catch 錯(cuò)誤的處理，而且支持你 yield 一個(gè)數(shù)組，對象，通過 promise.all 去實(shí)現(xiàn)。另外 yield thunk 函數(shù)的時(shí)候，它統(tǒng)一轉(zhuǎn)成 promise 去處理了。感興趣的可以去看一下 co，相信現(xiàn)在一定很明朗了。
async/await
現(xiàn)在 JS 中用的最常用的異步解決方案了，不過 async 也是基于 generator 的實(shí)現(xiàn)，只不過是做了封裝。如果把 async/await 轉(zhuǎn)化成 generate/yield，只需要把 await 語法換成 yield，再扔到一個(gè) generate 函數(shù)中，async 的執(zhí)行換成 coPromise(gennerate) 就好了。
const asyncPromise = (x) => {
return new Promise((resolve) => {
setTimeout(() => {
resolve(x + 1);
}, 1000)
})
}
async function fn () {
const data = await asyncPromise(1);
console.log(data);
}
fn();
// 那轉(zhuǎn)化成 generator 可能就是這樣了。 coPromise 就是上面的實(shí)現(xiàn)
function* gen() {
const data = yield asyncPromise(1);
console.log(data);
}
coPromise(gen);復(fù)制代碼asyncToGenerator 就是這樣的原理，事實(shí)上 babel 也是這樣轉(zhuǎn)化的。
最后
我首先是通過 express 的中間件思想，實(shí)現(xiàn)了一個(gè) JS 中需求常見的 queue （異步隊(duì)列解決方案），然后再接著去實(shí)現(xiàn)一個(gè)簡單的 coThunk，最后把 thunk 換成 promise。因?yàn)楫惒浇鉀Q方案在 JS 中是很重要的，去使用現(xiàn)成的解決方案的時(shí)候，如果能去深入思考一下實(shí)現(xiàn)的原理，我相信是有助于我們學(xué)習(xí)進(jìn)步的。
