[Node.js] Promise，Q及Async

原文地址：http://www.moye.me/2014/12/27/promise_q_async/

引子

在使用Node/JS編程的時候，經常會遇到這樣的問題：有一連串的異步方法，需要按順序執行，前后結果之間有依賴關系，形如（片斷1）：

asyncTask(initial, function (err, result) {//step 1
    if (err)
        throw err;
    asyncTask(result, function (err, result2) {//step 2
        if (err)
            throw err;
        asyncTask(result2, function (err, result3) {//final
            if (err)
                throw err;
            console.log(result3);
        });
    });
});

 之前也介紹過，這就是著名的回調地獄（Pyramid of Doom）。

Promise

解決回調嵌套及串行狀態傳送問題，是有規范可循的，如 CommonJS的Promise規范 。

Promise是對異步編程的一種抽象。它是一個代理對象，代表一個必須進行異步處理的函數返回的值或拋出的異常。

以實現較多的 Promise/A（thenable）來說：

• Promise作為一個對象，代表了一次任務的執行，它有三種狀態：成功/失敗/執行中
• Promise對象有一個 then接口（原型函數），形如：then(fulfilledHandler, errorHandler, progressHandler)。這三個參數在Promise對象完成任務時被執行，它們對應狀態為成功/失敗/執行中——我們可以把then方法等同於Promise對象的構造器，fulfilledHandler、errorHandler及progressHandler可以對應到Promise.resolve(val)、Promise.reject(reason)及Promise.notify(update)
• Promise對象還有一個catch接口（原型函數），形如：catch(onRejected)。onRejected為錯誤處理的回調，接收從Promise.reject(reason)傳遞過來的錯誤信息
• Promise對象暴露給調用方的then接口，是一個永遠可以返回Promise對象自身的函數，所以then可以繼續鏈式調用then，直到任務完成——簡單說，Promise執行的結果可以傳遞給下一個Promise對象，這在異步任務的串行化中非常有用（並且我們不用擔心這些Promise在執行任務時產生副作用，根據規范，每一個Promise與被調函數間都是相互獨立的
• Promise對象的內部，維護一個隊列，在構造執行鏈完成時，將待執行函數存入需要依次執行的任務——什么時候算完成呢？一般的Promise庫實現，都需要在構造鏈的尾部調用一個done之類的函數，以明示構造鏈的終結。

Promise做為類和對象的細節，MDN上的描述更為詳盡。回到實現層面，先來看一個開源的Promise框架，Q：

Q

Q是一個對Promise/A規范實現較為完備的開源框架。

針對前述的代碼片斷1 場景，Q提供了這樣的可能性：Q.promise能將異步邏輯包裝成一個thenable函數，從而注入它實現的回調函數，源碼形如：

Q.promise = promise;
function promise(resolver) {
    if (typeof resolver !== "function") {
        throw new TypeError("resolver must be a function.");
    }
    var deferred = defer();
    try {
        resolver(deferred.resolve, 
            deferred.reject, deferred.notify);
    } catch (reason) {
        deferred.reject(reason);
    }
    return deferred.promise;
}

 這個resolver就是我們的異步邏輯封裝函數， 我們可以選擇性的接收resolve和reject作為參數，並在異步方法完成/出錯時進行回調，讓Q獲得流程控制權。

Q的異步串行例子

假設有兩個文本文件：1.txt 和 2.txt，內容分別為：I'm the first text.\n和I'm the second text.\n。我們需要順序且異步的讀取它們，在全部讀取完成/出錯時，顯示相應信息。

首先，需要將異步方法進行包裝（片斷2）：

var Q = require('q');
var path = require('path');
var fs = require('fs');

function readFile(previous, fileName) {
    return Q.promise(function (resolve, reject) {
        fs.readFile(path.join(process.cwd(), fileName),
            function (error, text) {
                if (error) {
                    reject(new Error(error));
                } else {
                    resolve(previous + text.toString());
                }
            });
    });
}

fs.readFile 讀取文件，成功調用Q.defer.resolve，出錯調用Q.defer.reject。readFile做為用於串聯Promise對象的方法，提供了一個previous狀態參數，用於累加上次執行的結果。有了這個方法，基於then的串行邏輯就能這樣實現：

readFile('', '1.txt')
    .then(function (previous) {
        return readFile(previous, '2.txt');
    })
    .then(function (finalText) {
        console.log(finalText);
    })
    .catch(function (error) {
        console.log(error);
    })
    .done();

可以看出，thenable函數的鏈式調用總是能將上一個Promise.resolve的結果做為參數傳入。

Async

Async 嚴格說起來不是一個Promise的實現，而是一個異步工具集（utilities），通過源碼我們能看得很清楚，它導出了非常多的方法，集合/流程控制 都有涉及。

針對前述的代碼片斷1 場景，Async提供了若干種方法，挑兩個有代表性的：

Async.waterfall

waterfall形如waterfall(tasks, [callback])，tasks是一個function數組，[callback]參數是最終結果的回調。tasks數組里的函數按順序執行，當前任務可以接收上一個任務執行的結果，看個例子：

var async = require('async');
var path = require('path');
var fs = require('fs');

function readFile4WaterFall(previous, fileName, callback) {
    fs.readFile(path.join(process.cwd(), fileName),
        function (error, text) {
            callback(error, previous + text.toString());
        });
}
async.waterfall(
    [
        function (callback) {
            readFile4WaterFall('', '1.txt', callback)
        },
        function (previous, callback) {
            readFile4WaterFall(previous, '2.txt', callback);
        }
    ], function (err, result) {
        console.log(result);
    }
);

可以看出，不管是何種形式的異步流程控制，都需要注入實現的回調（這里是function(callback)），以獲取流程控制權。運行結果：

I'm the first text.
I'm the second text.

Async.series

series形如series(tasks, [callback])，和waterfall不同，tasks數組里的函數按順序執行，每個任務只接受一個callback注入，並不能傳遞上一次任務執行的結果。每一個函數執行的結果，都被push到了一個result數組里，也就是[callback(error, result)]的第二個參數。例子：

var async = require('async');
var path = require('path');
var fs = require('fs');

function readFile4Series(fileName, callback) {
    fs.readFile(path.join(process.cwd(), fileName),
        function (error, text) {
            callback(error, text.toString());
        });
}
async.series(
    [
        function (callback) {
            readFile4Series('1.txt', callback)
        },
        function (callback) {
            readFile4Series('2.txt', callback);
        }
    ], function (err, result) {
        console.log(result);
    }
);

運行結果：

[ 'I\'m the first text.\n', 'I\'m the second text.\n' ]

 

動態的異步串行

老實說，上面的示例僅僅展示了異步框架的威力，卻並不實用：在實踐中，我們遇到的情況並不是事先構造好要執行的函數鏈，而是代碼動態決定要執行哪些函數，即 .then 是動態拼接出來的。

以前示Q的片斷2 為例，要讀取的文件名存在數組里，我們需要針對文件名構造執行鏈：

//要讀取的文件數組
var files = ['1.txt', '2.txt', '3.txt'];
//要構造的Promise鏈
var tasks = [];

readFile高階函數需要稍加改造，因為不是顯式的構造鏈，原 .then 傳遞上次執行的函數需要嵌入到高階函數中：

function readFileDynamic(fileName) {
    return function(previous) { //.then callback
        return Q.promise(function (resolve, reject) {
            fs.readFile(path.join(process.cwd(), fileName),
                function (error, text) {
                    if (error) {
                        reject(new Error(error));
                    } else {
                        resolve(previous + text.toString());
                    }
                });
        });
    }
}

構造任務鏈和執行鏈：

files.forEach(function (fileName) {
    tasks.push(readFileDynamic(fileName));
});

var result = Q('');
tasks.forEach(function (f) {
    result = result.then(f);
});

調用：

result
    .then(function (finalText) {
        console.log(finalText);
    })
    .catch(function (error) {
        console.log(error);
    })
    .done();

如此，借助Q就可實現動態的異步串行鏈，本質和靜態構造執行鏈無二致，只是Promise構造形式進行了轉換。至於Async就更簡單了，前述的 waterfall/series的 tasks本就是個數組，天然動態。 

 

更多文章請移步我的blog新地址： http://www.moye.me/