async源碼之series

whidy 發(fā)布于2019-08-22 18:22 / 1171人閱讀

摘要：主線程從讀取回調(diào)函數(shù)并執(zhí)行。根據(jù)循環(huán)運(yùn)行數(shù)一次異步操作的最大數(shù)量，，進(jìn)入集合中第一個(gè)函數(shù)的調(diào)用，進(jìn)入，注冊(cè)回調(diào)函數(shù)。事件完成，回調(diào)函數(shù)進(jìn)入。采用同步功能并將其設(shè)置為異步，并將其返回值傳遞給回調(diào)函數(shù)。

前言

最近在看Node設(shè)計(jì)模式之異步編程的順序異步迭代，簡(jiǎn)單的實(shí)現(xiàn)如下：

function series(tasks, callback) {
    let results = [];
    function iterate(index) {
        if (index === tasks.length) {
            return finish();
        }
        const task = tasks[index];
        task(function(err, res) {
            results.push(res);
            iterate(index + 1);
        });
    }

    function finish() {
        // 迭代完成的操作
        callback(null, results);
    }

    iterate(0);
}

series(
    [
        callback => {
            setTimeout(function() {
                console.log(456);
                callback(null, 1);
            }, 500);
        },
        callback => {
            console.log(123);
            callback(null, 2);
        }
    ],
    function(err, results) {
        console.log(results);
    }
);

// 456
// 123
// [1, 2]

而async庫(kù)是一個(gè)非常流行的解決方案，在Node.js和JavaScript中來說，用于處理異步代碼。它提供了一組功能，可以大大簡(jiǎn)化不同配置中一組任務(wù)的執(zhí)行，并為異步處理集合提供了有用的幫助。

async庫(kù)可以在實(shí)現(xiàn)復(fù)雜的異步控制流程時(shí)大大幫助我們，但是一個(gè)難題就是選擇正確的庫(kù)來解決問題。例如，對(duì)于順序執(zhí)行，有大約20個(gè)不同的函數(shù)可供選擇。

好奇心起來，就想看看一個(gè)成熟的庫(kù)跟我們簡(jiǎn)單實(shí)現(xiàn)的代碼區(qū)別有多大。

series

按順序運(yùn)行任務(wù)集合中的函數(shù)，每個(gè)函數(shù)在前一個(gè)函數(shù)完成后運(yùn)行。如果系列中的任何函數(shù)將錯(cuò)誤傳遞給其回調(diào)函數(shù)，則不會(huì)運(yùn)行更多函數(shù)，并立即使用錯(cuò)誤值調(diào)用回調(diào)函數(shù)。否則，回調(diào)會(huì)在任務(wù)完成時(shí)收到一系列結(jié)果。

const async = require("async");

async.series({
    one: function(callback) {
        setTimeout(function() {
            callback(null, 1);
        }, 200);
    },
    two: function(callback){
        setTimeout(function() {
            callback(null, 2);
        }, 100);
    }
}, function(err, results) {
    console.log(results);
    // results is now equal to: {one: 1, two: 2}
});

我們來看看源碼，找到series方法，可以看到：

function series(tasks, callback) {
    _parallel(eachOfSeries, tasks, callback);
}

除了我們自己傳的兩個(gè)參數(shù)以外，默認(rèn)還傳了一個(gè)eachOfSeries，接著往下看：

function _parallel(eachfn, tasks, callback) {
    // noop：空的函數(shù)
    callback = callback || noop;
    // isArrayLike：檢查"value"是否與array相似
    var results = isArrayLike(tasks) ? [] : {};

    eachfn(tasks, function (task, key, callback) {
        // wrapAsync：包裝成異步
        wrapAsync(task)(function (err, result) {
            if (arguments.length > 2) {
                result = slice(arguments, 1);
            }
            results[key] = result;
            callback(err);
        });
    }, function (err) {
        callback(err, results);
    });
}

這里我們可以看到，_parallel方法其實(shí)就是eachOfSeries方法的調(diào)用。

先解釋一下eachOfSeries這三個(gè)參數(shù)：

第一個(gè)參數(shù)就是要執(zhí)行的函數(shù)的集合。

第二個(gè)參數(shù)可以看成每個(gè)函數(shù)的執(zhí)行（wrapAsync可以先忽略掉，直接看成這一個(gè)函數(shù)）。

第三個(gè)參數(shù)就是所有函數(shù)執(zhí)行完后的回調(diào)。

讓我們來看看eachOfSeries是如何的實(shí)現(xiàn)：

var eachOfSeries = doLimit(eachOfLimit, 1);

function eachOfLimit(coll, limit, iteratee, callback) {
    _eachOfLimit(limit)(coll, wrapAsync(iteratee), callback);
}

function doLimit(fn, limit) {
    return function (iterable, iteratee, callback) {
        return fn(iterable, limit, iteratee, callback);
    };
}

我們把上面進(jìn)行轉(zhuǎn)換，這樣看起來更明了些：

var eachOfSeries = function(iterable, iteratee, callback) {
    return _eachOfLimit(1)(iterable, wrapAsync(iteratee), callback);
};

Soga，最終就是調(diào)用_eachOfLimit完成的：

// limit：一次異步操作的最大數(shù)量，傳1可以看成串行，一個(gè)函數(shù)執(zhí)行完才進(jìn)行下一個(gè)
function _eachOfLimit(limit) {
    return function (obj, iteratee, callback) {
        // once：函數(shù)只運(yùn)行一次
        callback = once(callback || noop);
        if (limit <= 0 || !obj) {
            return callback(null);
        }
        // iterator：迭代器，有根據(jù)類型分類，這邊簡(jiǎn)單拿數(shù)組迭代器createArrayIterator來分析
        var nextElem = iterator(obj);
        var done = false;
        var running = 0;
        var looping = false;

        function iterateeCallback(err, value) {
            running -= 1;
            if (err) {
                done = true;
                callback(err);
            }
            else if (value === breakLoop || (done && running <= 0)) {
                done = true;
                return callback(null);
            }
            else if (!looping) {
                replenish();
            }
        }

        function replenish () {
            looping = true;
            while (running < limit && !done) {
                var elem = nextElem();
                if (elem === null) {
                    done = true;
                    if (running <= 0) {
                        callback(null);
                    }
                    return;
                }
                running += 1;
                // onlyOnce：函數(shù)只運(yùn)行一次
                iteratee(elem.value, elem.key, onlyOnce(iterateeCallback));
            }
            looping = false;
        }
        
        // 遞歸
        replenish();
    };
}

function once(fn) {
    return function() {
        if (fn === null) return;
        var callFn = fn;
        fn = null;
        callFn.apply(this, arguments);
    };
}

// 閉包大法，拿取集合中的函數(shù)
function createArrayIterator(coll) {
    var i = -1;
    var len = coll.length;
    return function next() {
        return ++i < len ? {value: coll[i], key: i} : null;
    }
}

終于，看到series的真身了。實(shí)現(xiàn)其實(shí)就是replenish()的遞歸大法。因?yàn)橐獙?shí)現(xiàn)串行，所以在replenish()中控制running數(shù)為1，取出集合中一個(gè)函數(shù)執(zhí)行，然后回調(diào)iterateeCallback()，running數(shù)減1，再調(diào)用replenish()，這樣就能控制每個(gè)函數(shù)在前一個(gè)函數(shù)完成后運(yùn)行。

說起來這流程還是比較簡(jiǎn)單，但是在異步編程里還是不太好理解，我們先來了解一下js執(zhí)行機(jī)制，再舉一個(gè)例子來看：

js執(zhí)行機(jī)制

同步的進(jìn)入主線程，異步的進(jìn)入Event Table并注冊(cè)函數(shù)。

當(dāng)指定的事情完成時(shí)，Event Table會(huì)將這個(gè)函數(shù)移入Event Queue。

主線程內(nèi)的任務(wù)執(zhí)行完畢為空，會(huì)去Event Queue讀取對(duì)應(yīng)的函數(shù)，進(jìn)入主線程執(zhí)行。

上述過程會(huì)不斷重復(fù)，也就是常說的Event Loop(事件循環(huán))。

普通版

function a() {
    setTimeout(function() {
        console.log(456);
    }, 500);
}

function b() {
    console.log(123);
}

function c() {
    setTimeout(function() {
        console.log(789);
    }, 0);
}

a();
b();
c();

// 123
// 789
// 456

按順序執(zhí)行可以看到

a()中setTimeout進(jìn)入Event Table，注冊(cè)回調(diào)函數(shù)。

b()，執(zhí)行console.log(123)。

c()中setTimeout進(jìn)入Event Table，注冊(cè)回調(diào)函數(shù)。

c()中setTimeout先完成，回調(diào)函數(shù)進(jìn)入Event Queue。

c()中setTimeout 500ms后完成，回調(diào)函數(shù)進(jìn)入Event Queue。

主線程從Event Queue讀取回調(diào)函數(shù)并執(zhí)行。

series版

const async = require("async");

async.series(
    [
        callback => {
            setTimeout(function() {
                console.log(456);
                callback(null, 1);
            }, 500);
        },
        callback => {
            console.log(123);
            callback(null, 2);
        },
        callback => {
            setTimeout(function() {
                console.log(789);
                callback(null, 3);
            }, 0);
        }
    ],
    function(err, results) {
        console.log(results);
    }
);

// 456
// 123
// 789
// [ 2, 1, 3 ]

按我自己的理解，主線程和Event Loop都執(zhí)行完稱為一輪：

第一輪

按照上面流程，主線程走到_eachOfLimit()，調(diào)用replenish()。根據(jù)while循環(huán)（運(yùn)行數(shù)running < 一次異步操作的最大數(shù)量 limit），running += 1，進(jìn)入集合中第一個(gè)函setTimeout數(shù)的調(diào)用，setTimeout進(jìn)入Event Table，注冊(cè)回調(diào)函數(shù)。

回到while循環(huán)，running=limit，結(jié)束循環(huán)，結(jié)束主線程。

setTimeout事件完成，回調(diào)函數(shù)進(jìn)入Event Queue。

主線程從Event Queue讀取回調(diào)函數(shù)并執(zhí)行，回調(diào)iterateeCallback，running -= 1，調(diào)用replenish()。

第二輪

重復(fù)第一輪。只要的區(qū)別在于集合中的第二個(gè)函數(shù)是同步的，所有是主線程一路執(zhí)行下來。

第三輪

重復(fù)第一輪。

第四輪

集合中的三個(gè)函數(shù)已經(jīng)都執(zhí)行完了，通過iterator()閉包拿到是null，回調(diào)最終結(jié)果。

wrapAsync

function wrapAsync(asyncFn) {
    return isAsync(asyncFn) ? asyncify(asyncFn) : asyncFn;
}

var supportsSymbol = typeof Symbol === "function";

function isAsync(fn) {
    return supportsSymbol && fn[Symbol.toStringTag] === "AsyncFunction";
}

wrapAsync()先判斷是否異步函數(shù)，如果是es7 Async Functions的話調(diào)用asyncify，否則返回原函數(shù)。

function asyncify(func) {
    return initialParams(function (args, callback) {
        var result;
        try {
            result = func.apply(this, args);
        } catch (e) {
            return callback(e);
        }
        // if result is Promise object
        if (isObject(result) && typeof result.then === "function") {
            result.then(function(value) {
                invokeCallback(callback, null, value);
            }, function(err) {
                invokeCallback(callback, err.message ? err : new Error(err));
            });
        } else {
            callback(null, result);
        }
    });
}

var initialParams = function (fn) {
    return function (/*...args, callback*/) {
        var args = slice(arguments);
        var callback = args.pop();
        fn.call(this, args, callback);
    };
};

采用同步功能并將其設(shè)置為異步，并將其返回值傳遞給回調(diào)函數(shù)。如果傳遞給asyncify的函數(shù)返回一個(gè)Promise，則該P(yáng)romise的resolved/rejected狀態(tài)將用于調(diào)用回調(diào)，而不僅僅是同步返回值。

總結(jié)

平日用慣async-await、promise，用起來簡(jiǎn)單，但也導(dǎo)致缺少思考。而嘗試用原生js去模擬，閱讀源碼，卻能帶來更多的收獲。

github地址，喜歡的支持star一下，Thanks?(?ω?)?。

GPU云服務(wù)器云服務(wù)器源碼中文說明之 Series Time Series Database async

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

轉(zhuǎn)載請(qǐng)注明本文地址：http://m.hztianpu.com/yun/95792.html

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

登陸后可評(píng)論

0條評(píng)論

whidy

男|高級(jí)講師

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

TA的文章

目前國(guó)內(nèi)有哪些免費(fèi)的云數(shù)據(jù)庫(kù)？

閱讀 2202·2021-11-22 13:52
牛啤~這個(gè)框架被大量使用，騰訊開源的RPC框架阿里的Dubbo全靠它

閱讀 1067·2021-11-17 09:33
騰訊云秒殺活動(dòng)：國(guó)內(nèi)vps云主機(jī),2核4G/80GB/8Mbps帶寬,79元/年;4核8G5M僅21

閱讀 2783·2021-09-01 10:49
前端實(shí)例練習(xí) - 模態(tài)相冊(cè)

閱讀 2908·2019-08-30 15:53
Flex布局

閱讀 2714·2019-08-29 16:10
float 實(shí)現(xiàn)三欄布局

閱讀 2483·2019-08-29 11:31
NPM酷庫(kù)039：iconv-lite，字符串編碼轉(zhuǎn)換

閱讀 1480·2019-08-26 11:40
Authing 和云原生應(yīng)用(CloudNative)

閱讀 1965·2019-08-26 10:59

成人无码视频,亚洲精品久久久久av无码,午夜精品久久久久久毛片,亚洲中文字幕日韩无码

資訊專欄INFORMATION COLUMN

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

async源碼之series

js執(zhí)行機(jī)制

wrapAsync

相關(guān)文章

NodeJs爬蟲抓取古代典籍，共計(jì)16000個(gè)頁面心得體會(huì)總結(jié)及項(xiàng)目分享

Nodejs異步流程框架async

**從“async”到async——Node異步流程控制總結(jié)**

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

Sails.js 內(nèi)存暴漲 & 源碼分析

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

0條評(píng)論

whidy

男|高級(jí)講師

TA的文章

目前國(guó)內(nèi)有哪些免費(fèi)的云數(shù)據(jù)庫(kù)？

牛啤~這個(gè)框架被大量使用，騰訊開源的RPC框架阿里的Dubbo全靠它

騰訊云秒殺活動(dòng)：國(guó)內(nèi)vps云主機(jī),2核4G/80GB/8Mbps帶寬,79元/年;4核8G5M僅21

前端實(shí)例練習(xí) - 模態(tài)相冊(cè)

Flex布局

float 實(shí)現(xiàn)三欄布局

NPM酷庫(kù)039：iconv-lite，字符串編碼轉(zhuǎn)換

Authing 和云原生應(yīng)用(CloudNative)

最新活動(dòng)

資訊專欄INFORMATION COLUMN

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

async源碼之series

js執(zhí)行機(jī)制

wrapAsync

相關(guān)文章

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

0條評(píng)論

男|高級(jí)講師

TA的文章

最新活動(dòng)

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