异步瀑布中带有 Mysql 连接池的 ForEach 函数

Posted 2023-03-07

【中文标题】异步瀑布中带有 Mysql 连接池的 ForEach 函数

【英文标题】：ForEach function with Mysql connection pool in Async waterfall

【发布时间】：2017-02-27 00:43:51

【问题描述】：

我已经关注this tutorial，现在我正在构建一个 Express js 应用程序，该应用程序连接到具有工作连接池的 mysql 数据库；然后我在我的服务器中使用 Async.js 作为指令瀑布。

我只是在重构我的代码以避免回调地狱。我很困惑，因为我有一个带有嵌套 forEach 函数的工作代码（回调地狱情况），它完美地获取了我的 MySql 连接，现在似乎没有任何工作。

问题： 在收到 GET 请求时，它会一直打印到“操作二”并且卡住。

这是我目前的MWE：

apiRoutes.get('/endpoint', function(req, res) 
async.waterfall([
function actionOne(callback)
    connection.acquire(function(err, con)
        con.query( myQuery , function(err, result)
            con.release();
            if(err)
                console.log(err);
                callback(err);
             else 
                console.log("Action One Success");
                callback(null, result);
            
        );
    );
,
function actionTwo(list, callback)
    console.log("Action Two");
    var arr = [];
    list.forEach(function(item, index, array)
        item.arr = [];
        connection.acquire(function(err, con)
            con.query( otherQuery , function(err, result)
                con.release();
                if(err)
                    console.log("SQL ERROR: "+err);
                    callback(err);
                 else 
                    item.arr = result;
                    arr.push(cult);
                    if(index === array.length-1)
                        console.log("Action Two Success");
                        callback(null, arr);
                    
                
            
        );
    )
,
function actionThree(item, callback)
         ....
    res.json('success');
],function(err)
     if(err)
      console.log(err);
);

---

已解决： 最后，我找到了使用 Promises 的链式异步指令的最佳可读解决方案。

apiRoutes.get('/endpoint', function(req, res)

    //Define 1st function
    function actionOne()
        return new Promise(function(fulfill, reject)
            myAsyncQueryFunction(err, result)
                if(err)  reject(err); 
                else  fulfill(result); 
            
        
    ;

    //Define 2nd function
    function actionTwo()
        return new Promise(function(fulfill, reject)
            actionOne().then(function(result)
                my2ndQueryFun(err, result)
                   if(err)  reject(err); 
                   else  fulfill(result); 
                
            , reject);
        
    ;

    //Execute async chained tasks, then send result to client
    actionTwo().then(function(result)
        res.json(result);
    ;

【参考方案1】：

如果由于某种原因，index === array.length-1 永远不会为真，您的代码将永远停留在actionTwo，因为它的回调函数永远不会被调用。

此外，您似乎正在根据您正在查询的数组的索引来控制何时调用 actionTwo 回调。但这可能会导致错误。 connection.acquire 和 con.query 都是异步的，因此位于最后一个索引并不能保证它是最后一个返回的查询。如果对列表最后一项的查询是运行最快的呢？

考虑先声明所有 SQL 查询任务，然后使用另一个 async.js 函数（如 async.parallel 或 async.series）控制它们的流程

function actionTwo(list, callback)
    console.log("Action Two");


    var tasks = [];

    list.forEach(function(item, index, array)
        item.arr = [];
        tasks.push(async.apply(getDataFromSQL, item));
    )

    // could also be async.series(tasks, function (err, results) 
    // it depends if you want to run the queries in parallel or not
    async.parallel(tasks, function (err, results) 
        if (err) 
            console.log("Action Two Error")
            return callback(err);
        
        console.log("Action Two Success");
        return callback(null, results); //results is an array of the "cult" objects returned at each getDataFromSQL callback
    );


    function getDataFromSQL(item, sqlCallback) 
        // otherQuery = generate_query_based_on_item(item);
        connection.acquire(function(err, con)
            con.query( otherQuery , function(err, result)
                con.release();
                if(err)
                    console.log("SQL ERROR: "+err);
                    sqlCallback(err);
                 else 
                    item.arr = result;
                    sqlCallback(null, cult); //not sure what cult meant. maybe result?
                
            
        );
    

【讨论】：

已解决：问题是异步方法connection.acquire和con.query，第一次调用没有问题，但第二次没有时间运行，因为整个方法在第二次之前返回SQL 结果。非常感谢！

指使用其他异步函数。我选择使用 async.waterfall 是因为每个函数都需要将他的结果传递给下一个函数。看here