Async詳解之一：流程控制

爲了適應異步編程，減小回調的嵌套，我嘗試了不少庫。最終以爲仍是async最靠譜。

地址：https://github.com/caolan/async

Async的內容分爲三部分：

1. 流程控制：簡化十種常見流程的處理

2. 集合處理：如何使用異步操做處理集合中的數據

3. 工具類：幾個經常使用的工具類

本文介紹其中最簡單最經常使用的流程控制部分。

因爲nodejs是異步編程模型，有一些在同步編程中很容易作到的事情，如今卻變得很麻煩。Async的流程控制就是爲了簡化這些操做。

1. series(tasks, [callback]) （多個函數依次執行，之間沒有數據交換）

有多個異步函數須要依次調用，一個完成以後才能執行下一個。各函數之間沒有數據的交換，僅僅須要保證其執行順序。這時可以使用series。

純js代碼：

step1(function(err, v1) {

  step2(function(err, v2) {

    step3(function(err, v3) {

       // do somethig with the err or values v1/v2/v3

    }

  }

});

從中能夠看到這嵌套仍是比較多深的，若是再多幾步，會更深。在代碼中忽略對了每一層err的處理，不然還都等加上 if(err) return callback(err)，那就更麻煩了。

對於這種狀況，使用async來處理，就是這樣的：

var async = require(‘async’)

async.series([

   step1, step2, step3

], function(err, values) {

   // do somethig with the err or values v1/v2/v3

});

能夠看到代碼簡潔了不少，並且自動處理每一個回調中的錯誤。固然，這裏只給出來最最簡單的例子，在實際中，咱們常會在每一個step中執行一些操做，這時可寫成：

var async = require(‘async’)

async.series([

  function(cb) { step1(function(err,v1) {

     // do something with v1

     cb(err, v1);

  }),

  function(cb) { step2(...) },

  function(cb) { step3(...) }

], function(err, values) {

// do somethig with the err or values v1/v2/v3

});

該函數的詳細解釋爲：

1. 依次執行一個函數數組中的每一個函數，每個函數執行完成以後才能執行下一個函數。

2. 若是任何一個函數向它的回調函數中傳了一個error，則後面的函數都不會被執行，而且將會馬上會將該error以及已經執行了的函數的結果，傳給series中最後那個callback。

3. 當全部的函數執行完後（沒有出錯），則會把每一個函數傳給其回調函數的結果合併爲一個數組，傳給series最後的那個callback。

4. 還能夠json的形式來提供tasks。每個屬性都會被看成函數來執行，而且結果也會以json形式傳給series最後的那個callback。這種方式可讀性更高一些。

具體例子可參考：https://github.com/freewind/async_demo/blob/master/series.js

其代碼中還包含了：

1. 若是中間某個函數出錯，series函數如何處理

2. 若是某個函數傳給回調的值爲undefined, null, {}, []等，series如何處理

另外還須要注意的是：多個series調用之間是不分前後的，由於series自己也是異步調用。

2. parallel(tasks, [callback]) （多個函數並行執行）

並行執行多個函數，每一個函數都是當即執行，不須要等待其它函數先執行。傳給最終callback的數組中的數據按照tasks中聲明的順序，而不是執行完成的順序。

若是某個函數出錯，則馬上將err和已經執行完的函數的結果值傳給parallel最終的callback。其它未執行完的函數的值不會傳到最終數據，但要佔個位置。

同時支持json形式的tasks，其最終callback的結果也爲json形式。

示例代碼：

async.parallel([ 
    function(cb) { t.fire('a400', cb, 400) }, 
    function(cb) { t.fire('a200', cb, 200) }, 
    function(cb) { t.fire('a300', cb, 300) } 
], function (err, results) { 
    log(’1.1 err: ‘, err); // -> undefined 
    log(’1.1 results: ‘, results); // ->[ 'a400', 'a200', 'a300' ] 
});

 

中途出錯的示例：

async.parallel([ 
    function(cb) { log('1.2.1: ', 'start'); t.fire('a400', cb, 400) }, // 該函數的值不會傳給最終callback，但要佔個位置 
    function(cb) { log('1.2.2: ', 'start'); t.err('e200', cb, 200) }, 
    function(cb) { log('1.2.3: ', 'start'); t.fire('a100', cb, 100) } 
], function(err, results) { 
    log(’1.2 err: ‘, err); // -> e200 
    log(’1.2 results: ‘, results); // -> [ , undefined, 'a100' ] 
});

 

以json形式傳入tasks

async.parallel({ 
    a: function(cb) { t.fire(‘a400′, cb, 400) }, 
    b: function(cb) { t.fire(‘c300′, cb, 300) } 
}, function(err, results) { 
    log(’1.3 err: ‘, err); // -> undefined 
    log(’1.3 results: ‘, results); // -> { b: ‘c300′, a: ‘a400′ } 
});

更詳細示例參見：https://github.com/freewind/async_demo/blob/master/parallel.js

3. waterfall(tasks, [callback]) （多個函數依次執行，且前一個的輸出爲後一個的輸入）

與seires類似，按順序依次執行多個函數。不一樣之處，每個函數產生的值，都將傳給下一個函數。若是中途出錯，後面的函數將不會被執行。錯誤信息以及以前產生的結果，將傳給waterfall最終的callback。

這個函數名爲waterfall(瀑布)，能夠想像瀑布從上到下，中途衝過一層層突起的石頭。注意，該函數不支持json格式的tasks。

async.waterfall([ 
    function(cb) { log('1.1.1: ', 'start'); cb(null, 3); }, 
    function(n, cb) { log('1.1.2: ',n); t.inc(n, cb); }, 
    function(n, cb) { log('1.1.3: ',n); t.fire(n*n, cb); } 
], function (err, result) { 
    log(’1.1 err: ‘, err); // -> null 
    log(’1.1 result: ‘, result); // -> 16 
});

更詳細示例參見：https://github.com/freewind/async_demo/blob/master/waterfall.js

4. auto(tasks, [callback]) （多個函數有依賴關係，有的並行執行，有的依次執行）

用來處理有依賴關係的多個任務的執行。好比某些任務之間彼此獨立，能夠並行執行；但某些任務依賴於其它某些任務，只能等那些任務完成後才能執行。

雖然咱們可使用async.parallel和async.series結合起來實現該功能，但若是任務之間關係複雜，則代碼會至關複雜，之後若是想添加一個新任務，也會很麻煩。這時使用async.auto，則會事半功倍。

若是有任務中途出錯，則會把該錯誤傳給最終callback，全部任務（包括已經執行完的）產生的數據將被忽略。

這裏假設我要寫一個程序，它要完成如下幾件事：

1. 從某處取得數據

2. 在硬盤上創建一個新的目錄

3. 將數據寫入到目錄下某文件

4. 發送郵件，將文件以附件形式發送給其它人。

分析該任務，能夠知道1與2能夠並行執行，3須要等1和2完成，4要等3完成。

async.auto({ 
    getData: function (callback) { 
        setTimeout(function(){ 
            console.log(’1.1: got data’); 
            callback(); 
        }, 300); 
    }, 
    makeFolder: function (callback) { 
        setTimeout(function(){ 
            console.log(’1.1: made folder’); 
            callback(); 
        }, 200); 
    }, 
    writeFile: ['getData', 'makeFolder', function(callback) { 
        setTimeout(function(){ 
            console.log('1.1: wrote file'); 
            callback(null, 'myfile'); 
        }, 300); 
    }], 
    emailFiles: ['writeFile', function(callback, results) { 
        log('1.1: emailed file: ', results.writeFile); // -> myfile 
        callback(null, results.writeFile); 
    }] 
}, function(err, results) { 
    log(’1.1: err: ‘, err); // -> null 
    log(’1.1: results: ‘, results); // -> { makeFolder: undefined, 
                                    //      getData: undefined, 
                                    //      writeFile: ‘myfile’, 
                                    //      emailFiles: ‘myfile’ } 
});

更多詳細示例參見：https://github.com/freewind/async_demo/blob/master/auto.js

5. whilst(test, fn, callback)（用可於異步調用的while）

至關於while，但其中的異步調用將在完成後纔會進行下一次循環。舉例以下：

var count1 = 0; 
async.whilst( 
    function() { return count1 < 3 }, 
    function(cb) { 
        log(’1.1 count: ‘, count1); 
        count1++; 
        setTimeout(cb, 1000); 
    }, 
    function(err) { 
        // 3s have passed 
        log(’1.1 err: ‘, err); // -> undefined 
    } 
);

它至關於：

try { 
  whilst(test) { 
    fn(); 
  } 
  callback(); 
} catch (err) { 
  callback(err); 
}

該函數的功能比較簡單，條件變量一般定義在外面，可供每一個函數訪問。在循環中，異步調用時產生的值實際上被丟棄了，由於最後那個callback只能傳入錯誤信息。

另外，第二個函數fn須要能接受一個函數cb，這個cb最終必須被執行，用於表示出錯或正常結束。

更詳細示例參見：https://github.com/freewind/async_demo/blob/master/whilst_until.js

6. until(test, fn, callback) （與while類似，但判斷條件相反）

var count4 = 0; 
async.until( 
    function() { return count4>3 }, 
    function(cb) { 
        log(’1.4 count: ‘, count4); 
        count4++; 
        setTimeout(cb, 200); 
    }, 
    function(err) { 
        // 4s have passed 
        log(’1.4 err: ‘,err); // -> undefined 
    } 
);

當第一個函數條件爲false時，繼續執行第二個函數，不然跳出。

7. queue （可設定worker數量的隊列）

queue至關於一個增強版的parallel，主要是限制了worker數量，再也不一次性所有執行。當worker數量不夠用時，新加入的任務將會排隊等候，直到有新的worker可用。

該函數有多個點可供回調，如worker用完時、無等候任務時、所有執行完時等。

定義一個queue，其worker數量爲2，並在任務執行時，記錄一下日誌：

var q = async.queue(function(task, callback) { 
    log(‘worker is processing task: ‘, task.name); 
    task.run(callback); 
}, 2);

worker數量將用完時，會調用saturated函數：

q.saturated = function() { 
    log(‘all workers to be used’); 
}

當最後一個任務交給worker執行時，會調用empty函數

q.empty = function() { 
    log(‘no more tasks wating’); 
}

當全部任務都執行完時，會調用drain函數

q.drain = function() { 
    console.log(‘all tasks have been processed’); 
}

放入多個任務，可一次放一個，或一次放多個

q.push({name:’t1′, run: function(cb){ 
    log(‘t1 is running, waiting tasks: ‘, q.length()); 
    t.fire(‘t1′, cb, 400); // 400ms後執行 
}}, function(err) { 
    log(‘t1 executed’); 
});

 

q.push([{name:'t3', run: function(cb){ 
    log('t3 is running, waiting tasks: ', q.length()); 
    t.fire('t3', cb, 300); // 300ms後執行 
}},{name:'t4', run: function(cb){ 
    log('t4 is running, waiting tasks: ', q.length()); 
    t.fire('t4', cb, 500); // 500ms後執行 
}}], function(err) { 
    log(‘t3/4 executed’); 
});

 

更多詳細示例參見：https://github.com/freewind/async_demo/blob/master/queue.js

8. iterator(tasks) （將幾個函數包裝爲iterator）

將一組函數包裝成爲一個iterator，可經過next()獲得如下一個函數爲起點的新的iterator。該函數一般由async在內部使用，但若是須要時，也可在咱們的代碼中使用它。

var iter = async.iterator([ 
function() { console.log('111') }, 
function() { console.log('222') }, 
function() { console.log('333') } 
]);

console.log(iter());

console.log(iter.next());

直接調用()，會執行當前函數，並返回一個由下個函數爲起點的新的iterator。調用next()，不會執行當前函數，直接返回由下個函數爲起點的新iterator。

對於同一個iterator，屢次調用next()，不會影響本身。若是隻剩下一個元素，調用next()會返回null。

更詳細示例參見：https://github.com/freewind/async_demo/blob/master/iterator.js

9. apply(function, arguments..) （給函數預綁定參數）

apply是一個很是好用的函數，可讓咱們給一個函數預綁定多個參數並生成一個可直接調用的新函數，簡化代碼。

對於函數：

function(callback) { t.inc(3, callback); }

能夠用apply改寫爲：

async.apply(t.inc, 3);

還能夠給某些函數預設值，獲得一個新函數：

var log = async.apply(console.log, ">");

log(‘hello’);

// > hello

更詳細代碼參見：https://github.com/freewind/async_demo/blob/master/apply.js

10. nextTick(callback) （在nodejs與瀏覽器兩邊行爲一致）

nextTick的做用與nodejs的nextTick同樣，都是把某個函數調用放在隊列的尾部。但在瀏覽器端，只能使用setTimeout(callback,0)，但這個方法有時候會讓其它高優先級的任務插到前面去。

因此提供了這個nextTick，讓一樣的代碼在服務器端和瀏覽器端表現一致。

var calls = [];

async.nextTick(function() {

    calls.push(‘two’);

});

calls.push(‘one’);

async.nextTick(function() {

    console.log(calls); // -> [ 'one', 'two' ]

});

更詳細代碼參見：https://github.com/freewind/async_demo/blob/master/nextTick.js