JavaScript 異步隊(duì)列及Co實(shí)現(xiàn)

LdhAndroid 發(fā)布于2019-08-21 16:49 / 2312人閱讀

摘要：在中，又由于單線程的原因，異步編程又是非常重要的。方法有很多，，，觀察者，，，這些中處理異步編程的，都可以做到這種串行的需求。

引入

隊(duì)列對(duì)于任何語言來說都是重要的，io 的串行，請(qǐng)求的并行等等。在 JavaScript 中，又由于單線程的原因，異步編程又是非常重要的。昨天由一道面試題的啟發(fā)，我去實(shí)現(xiàn) JS 中的異步隊(duì)列的時(shí)候，借鑒了 express 中間件思想，并發(fā)散到 co 實(shí)現(xiàn) 與 generator，以及 asyncToGenerator。

本次用例代碼都在此，可以 clone 下來試一下

異步隊(duì)列

很多面試的時(shí)候會(huì)問一個(gè)問題，就是怎么讓異步函數(shù)可以順序執(zhí)行。方法有很多，callback，promise，觀察者，generator，async/await，這些 JS 中處理異步編程的，都可以做到這種串行的需求。但是很麻煩的是，處理起來是挺麻煩的，你要不停的手動(dòng)在上一個(gè)任務(wù)調(diào)用下一個(gè)任務(wù)。比如 promise，像這樣：

a.then(() => b.then(() => c.then(...)))

代碼嵌套的問題，有點(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ì)列才會(huì)繼續(xù)往下走。那這個(gè) next 就至關(guān)重要了，它會(huì)控制隊(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è)對(duì)象
  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è)。

我這里沒去構(gòu)造一個(gè) class，而是通過閉包的特性去處理的。queue 方法返回一個(gè)包含 add，run 的對(duì)象，add 即為像隊(duì)列中添加異步方法，run 就是開始執(zhí)行。在 queue 內(nèi)部，我們定義了幾個(gè)變量，list 用來保存隊(duì)列，index 就是游標(biāo)，表示隊(duì)列現(xiàn)在走到哪個(gè)函數(shù)了，另外，最重要的是 next 方法，它是控制游標(biāo)向后移動(dòng)的。

run 函數(shù)一旦執(zhí)行，隊(duì)列即開始 run。一開始執(zhí)行隊(duì)列里的第一個(gè) async 函數(shù)，我們把 next 函數(shù)傳給了它，然后由 async 函數(shù)決定什么時(shí)候執(zhí)行 next，即開始執(zhí)行下一個(gè)任務(wù)。我們沒有并不知道異步任務(wù)什么時(shí)候才算完成，只能通過打成某種共識(shí)，來告知 queue 某個(gè)任務(wù)完成。就是傳給任務(wù)的 next 函數(shù)。其實(shí) async 返回的這個(gè)函數(shù)，有一個(gè)名字，叫 Thunk，后面我們會(huì)簡單介紹。

Thunk

thunk 其實(shí)是為了解決 “傳名調(diào)用” 的。就是我傳給函數(shù) A 一個(gè)表達(dá)式作參數(shù) x + 1，但是我不確定這個(gè) x + 1 什么時(shí)候會(huì)用到，以及會(huì)不會(huì)用到，如果在傳入就執(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;
}

嗯... 其實(shí)就是一個(gè)回調(diào)函數(shù)...

暫停

其實(shí)只要某個(gè)任務(wù)，不繼續(xù)調(diào)用 next，隊(duì)列就已經(jīng)不會(huì)繼續(xù)往下走了。比如我們 async 任務(wù)里加一個(gè)判斷（通常是異步 io，請(qǐng)求的容錯(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

當(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 的好處在于，你可以主動(dòng)的停止隊(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 jump = () => {
    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)

其實(shí)還是加攔截... 只不過從 async 函數(shù)中，換到了 next 函數(shù)里面，利用 isStop 這個(gè)變量切換 true/false，開關(guān)暫停。我加了兩個(gè)定時(shí)器，一個(gè)是 3 秒后暫停，一個(gè)是 5 秒后繼續(xù)，（請(qǐng)忽略定時(shí)器的誤差），按道理應(yīng)該是隊(duì)列到三秒的時(shí)候，也就是第三個(gè)任務(wù)執(zhí)行完暫停，然后再隔 2 秒，繼續(xù)。結(jié)果打印到 3 的時(shí)候，停住，兩秒之后繼續(xù) 4，5，6.

兩種思路，請(qǐng)結(jié)合場(chǎng)景思考問題。

并發(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

我添加了一個(gè) parallelRun 方法，用于并行，我覺得還是不要放到 run 函數(shù)里面了，抽象單元盡量細(xì)化還是。然后還加了一個(gè) isParallel 的變量，默認(rèn)是 false，考慮到 next 函數(shù)有可能會(huì)被調(diào)用，所以需要加一個(gè)攔截，保證不會(huì)處亂。

以上就是利用僅用 thunk 函數(shù)，結(jié)合 next 實(shí)現(xiàn)的異步隊(duì)列控制器，queue，跟你可以把 es6 代碼都改成 es5，保證兼容，當(dāng)然是足夠簡單的，不適用于負(fù)責(zé)的場(chǎng)景 ?，僅提供思路。

generator 與 co

為什么要介紹 generator，首先它也是用來解決異步回調(diào)的，另外它的使用方式也是調(diào)用 next 函數(shù)，generator 才會(huì)往下執(zhí)行，默認(rèn)是暫停狀態(tài)。yield 就相當(dāng)于上面的 q.add，往隊(duì)列中添加任務(wù)。所以我也打算一起介紹，來更好的拓寬思路。發(fā)散思維，相似的知識(shí)點(diǎn)做好歸納，然后某一天你就會(huì)突然有一種：原來是這么回事，原來 xxx 是借鑒子 yyy，然后你又去研究 yyy - -。

簡介 generator

簡單介紹回顧一下，因?yàn)橛型瑢W(xué)不經(jīng)常用，肯定會(huì)有遺忘。

// 一個(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}

首先生成器其實(shí)就是一個(gè)通過函數(shù)體內(nèi)部定義迭代算法，然后返回一個(gè) iterator 對(duì)象。關(guān)于iterator，可以看我另一篇文章。
gen 執(zhí)行返回一個(gè)對(duì)象 g，而不是返回結(jié)果。g 跟其他 iterator 一樣，通過調(diào)用 next 方法，保證游標(biāo) + 1，并且返回一個(gè)對(duì)象，包含了 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è)弊端就是不會(huì)自動(dòng)執(zhí)行，TJ 大神寫了一個(gè) co，來自動(dòng)執(zhí)行 generator，也就是自動(dòng)調(diào)用 next。它要求 yield 后面的函數(shù)/語句，必須是 thunk 函數(shù)或者是 promise 對(duì)象，因?yàn)橹挥羞@樣才會(huì)串聯(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ù)會(huì)接收 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

這里定義的 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 對(duì)象的話，我們可以在 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 對(duì)象中的 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

其實(shí) co 的源碼就是通過這兩種思路實(shí)現(xiàn)的，只不過它做了更多的 catch 錯(cuò)誤的處理，而且支持你 yield 一個(gè)數(shù)組，對(duì)象，通過 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);

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)步的。

歡迎 star 個(gè)人 blog：https://github.com/sunyongjia... ?

GPU云服務(wù)器云服務(wù)器 javascript實(shí)現(xiàn)隊(duì)列異步隊(duì)列異步任務(wù)隊(duì)列異步消息隊(duì)列java

文章版權(quán)歸作者所有，未經(jīng)允許請(qǐng)勿轉(zhuǎn)載,若此文章存在違規(guī)行為，您可以聯(lián)系管理員刪除。

轉(zhuǎn)載請(qǐng)注明本文地址：http://www.ezyhdfw.cn/yun/88768.html

發(fā)表評(píng)論

登陸后可評(píng)論

0條評(píng)論

LdhAndroid

男|高級(jí)講師

我要關(guān)注我要私信

TA的文章

Java開發(fā)編程的好處在哪

閱讀 3746·2021-11-22 15:24
一根煙的時(shí)間帶你了解，程序員被稱為2021新時(shí)代農(nóng)民工如何改變外界的看法...

閱讀 1678·2021-09-26 09:46
PIGYun中秋特惠：香港/韓國VPS月付14元起

閱讀 1979·2021-09-14 18:01
【css3學(xué)習(xí)系列】之box-shadow,radial-gradient,linear-gradi

閱讀 2681·2019-08-30 15:45
關(guān)于CSS你應(yīng)該知道的基礎(chǔ)知識(shí) - 選擇器篇

閱讀 3579·2019-08-30 14:23
webview中嵌入部分html5適配的小方法~

閱讀 1950·2019-08-30 12:43
前端每日實(shí)戰(zhàn) 167# 視頻演示如何用 1 個(gè) dom 元素創(chuàng)作兩顆愛心

閱讀 3006·2019-08-30 10:56
webpack1學(xué)習(xí)筆記

閱讀 870·2019-08-29 12:20

亚洲中字慕日产2020,大陆极品少妇内射AAAAAA,无码av大香线蕉伊人久久,久久精品国产亚洲av麻豆网站

資訊專欄INFORMATION COLUMN

上云采購季！| 2核2G4M爆款云服務(wù)器低至59元/年，更有多臺(tái)、長期優(yōu)惠，快來選購！

JavaScript 異步隊(duì)列及Co實(shí)現(xiàn)

相關(guān)文章

Node.js 異步異聞錄

**總結(jié)javascript基礎(chǔ)概念（二）：事件隊(duì)列循環(huán)**

javascript 異步編程

《Node.js設(shè)計(jì)模式》基于ES2015+的回調(diào)控制流

**夯實(shí)基礎(chǔ)-JavaScript異步編程**

發(fā)表評(píng)論

0條評(píng)論

LdhAndroid

男|高級(jí)講師

TA的文章

Java開發(fā)編程的好處在哪

一根煙的時(shí)間帶你了解，程序員被稱為2021新時(shí)代農(nóng)民工如何改變外界的看法...

PIGYun中秋特惠：香港/韓國VPS月付14元起

【css3學(xué)習(xí)系列】之box-shadow,radial-gradient,linear-gradi

關(guān)于CSS你應(yīng)該知道的基礎(chǔ)知識(shí) - 選擇器篇

webview中嵌入部分html5適配的小方法~

前端每日實(shí)戰(zhàn) 167# 視頻演示如何用 1 個(gè) dom 元素創(chuàng)作兩顆愛心

webpack1學(xué)習(xí)筆記

最新活動(dòng)

資訊專欄INFORMATION COLUMN

上云采購季！| 2核2G4M爆款云服務(wù)器低至59元/年，更有多臺(tái)、長期優(yōu)惠，快來選購！

JavaScript 異步隊(duì)列及Co實(shí)現(xiàn)

相關(guān)文章

發(fā)表評(píng)論

0條評(píng)論

男|高級(jí)講師

TA的文章

最新活動(dòng)

上云采購季！| 2核2G4M爆款云服務(wù)器低至59元/年，更有多臺(tái)、長期優(yōu)惠，快來選購！