如何使用映射函数控制串行和并行控制流？答案

【问题标题】：How to control serial and parallel control flow with mapped functions?如何使用映射函数控制串行和并行控制流？
【发布时间】：2015-10-26 15:27:36
【问题描述】：

我画了一个简单的流程图，它基本上是从互联网上抓取一些数据并将它们加载到数据库中。到目前为止，我以为我对承诺很平静，但是现在我有一个问题，我至少工作了三天，没有一个简单的步骤。

这是流程图：

考虑有一个像这样的静态字符串数组：const courseCodes = ["ATA, "AKM", "BLG",... ]。

我有一个fetch 函数，它基本上执行一个HTTP 请求，然后进行解析。然后它返回一些对象数组。

fetch 可以完美地使用预期的对象数组调用其回调，它甚至可以与更大更整洁的 Promises 一起使用。

fetch 函数应该以courseCodes 数组中的每个元素作为其参数来调用。此任务应并行执行，因为这些单独的 fetch 函数不会相互影响。

因此，回调中应该有一个results 数组（或Promises 解析参数），其中包含对象数组的数组。有了这些结果，我应该使用 results 数组中的这些对象作为其参数来调用我的 loadCourse。这些任务应该在串行执行中执行，因为它基本上查询数据库是否存在类似对象，如果不存在则添加它。

如何在 node.js 中执行此类任务？在这种情况下，我无法维护异步流程。我在 caolan/async 库和 bluebird 和 q promise 库上失败了。

【问题讨论】：

标签： node.js asynchronous concurrency

【解决方案1】：

如果你能理解，试试这样的：

const courseCodes = ["ATA, "AKM", "BLG",... ]

//stores the tasks to be performed.
var parallelTasks = [];
var serialTasks = [];

//keeps track of courses fetched & results.
var courseFetchCount = 0;
var results = {};


//your fetch function.
fetch(course_code){
    //your code to fetch & parse.

    //store result for each course in results object
    results[course_code] = 'whatever result comes from your fetch & parse code...';
}

//your load function.
function loadCourse(results) {
    for(var index in results) {
        var result = results[index]; //result for single course;
        var task = (
            function(result) {
                return function() {
                    saveToDB(result);
                }
            }
        )(result);

        serialTasks.push(task);
    }
    //execute serial tasks for saving results to database or whatever.
    var firstSerialTask = serialTasks.shift();
    nextInSerial(null, firstSerialTask);
}

//pseudo function to save a result to database.
function saveToDB(result) {
    //your code to store in db here.
}

//checks if fetch() is complete for all course codes in your array
//and then starts the serial tasks for saving results to database.
function CheckIfAllCoursesFetched() {
    courseFetchCount++;
    if(courseFetchCount == courseCodes.length) {
        //now process courses serially
        loadCourse(results);
    }
}

//helper function that executes tasks in serial fashion.
function nextInSerial(err, result) {
    if(err) throw Error(err.message);

    var nextSerialTask = serialTasks.shift();
    nextSerialTask(result);
}

//start executing parallel tasks for fetching.
for(var index in courseCode) {
    var course_code = courseCode[index];
    var task = (
        function(course_code) {
            return function() {
                fetch(course_code);
                CheckIfAllCoursesFetched();
            }
        }
    )(course_code);
    parallelTasks.push(task);

    for(var task_index in parallelTasks) {
        parallelTasks[task_index]();
    }
}

或者你可以参考nimble npm 模块。

【讨论】：