管理异步并在节点中执行多个MySQL查询

Question

我遇到了一些处理nodejs异步性质的问题。我正在使用nodejs-mysql连接器来获取数据。实质上，下面的代码正在执行一个查询来获取file_url并将信息发送到myApi。第二个查询应该在从api（inputId）返回的结果更新数据库后立即运行。但是，之前的工作并不是以异步的方式进行的（连接在处理所有内容之前关闭）。使用async模块执行以下操作的最佳方法是什么？

var mysql = require("mysql"); 
var async = require("async"); 
var Q = require('q'), 
    myApi = require('myApi')('xxxxx'); 

//DB 
var con = mysql.createConnection({ 
    host: "localhost", 
    user: "root", 
    password: "root", 
    database: "test" 
}); 


//1ST Query - Fetch Urls, then upload file to api 
con.query('SELECT file_url, file_id FROM myfiles', function(err, rows) { 
    if (err) throw err; 
    for (var i = 0; i < rows.length; i++) { 

     // File URL for upload 
     var fileUrl = rows[i].file_url, 
      createInputPromise, createEncodingProfilePromise; 
     var fileId = rows[i].file_id; 

     // Create myApi Input 
     createInputPromise = myApi.input.create(fileUrl); 

     //WHERE THE ACTION OCCURS! 
     Q.all([createInputPromise, createEncodingProfilePromise]).then(
      function(result) { 
       console.log('Successfully uploaded file: ' + result[0].url + ' inputId: ' + result[0].inputId); 

       //2ND Query - Save the inputId return by api for each file 
       con.query(
        'UPDATE myfiles SET myApi_input_id = ? WHERE file_id = ?', [result[0].inputId, fileId], 
        function(err, result) { 
         if (err) throw err; 
         console.log('Changed ' + result.changedRows + ' rows'); 
        } 
       ); 

      }, 
      function(error) { 
       console.log('Error while uploading file to api:', error); 
      } 
     ); 
    }; 

}); 

con.end(function(err) {}); 

Answer 1

很多这取决于你如何处理这些：createInputPromise，createEncodingProfilePromise。与这两者相关的更多代码将有助于获得良好的答案。你在代码中究竟遇到了这个问题？ console.log('Successfully uploaded file: ' + result[0].url + ' inputId: ' + result[0].inputId);是否输出到控制台，或之前是否有错误？

现在，你确定myApi.input.create(fileUrl)正在退货吗？您需要查看该API的文档以确保它是。如果没有，那么你可以使用默认的Q库创建一个：

var inputPromise = Q.defer(); 
 
var encodingProfilePromise = Q.defer(); 
 

 
inputPromise.resolve(myApi.input.create(fileUrl)); 
 
encodingProfilePromise.resolve(/*encoding profile code*/); 
 

 
Q.all([inputPromise.promise, encodingProfilePromise.promise]).then(function(result) { 
 
    /*blah blah blah*/ 
 
});

随着越来越多的信息，我可以给出一个更精确的答案。

Answer 2

而不是异步模块，我会以一种不同的方式看它。让我们用发电机。

let co = require('co'); 
let parallel = require('co-parallel'); 
let sequelize = require('sequelize'); // also read documentation to provide connection details (this is just pseudocode) 


let gen = function*() { 
    let inputId = yield fetch_file_url(); // this function should be generator or should return promise 
    let res = yield sequelize.query(`INSERT INTO... ${inputId} `); 
} 

co(function*() { 
    let files = ['f1', 'f2', 'f3']; 
    let gens = files.map(file => gen(file)); 
    let arrayOrResults = yield parallel(gens, 5); // 5 threads 
    // done 
}).catch(function(err) { 
    // catch errors 
}); 

Answer 3

发布的答案是伟大的选择，但我最终想要得到它的成就与使用async。我遇到了waterfall模式：

运行的功能系列的阵列，每个路过的结果 接下来在数组中。但是，如果任何函数将错误 传递给回调，则不会执行下一个函数，并立即调用主要的 回调并返回错误。

我最终实现了以下内容：

var pool = mysql.createPool({...}); 

function fn(callback) { 
    var getConnection = function(callback) {...}; 

    async.waterfall([ 
     doQuery.bind(null, getConnection), 
     doSomethingAsyncWithResult, 
     doUpdate(null, getConnection) 
    ], function(err, result) { 
     getConnection.end(); 
     callback(err, result); 
    }); 

} 

function doQuery(connection, callback) { 
    connection.query(sql, callback); 
} 

function doSomethingAsyncWithResult(result, callback) { 
    ...something... 
    callback(null, anotherResult); 
} 

function doUpdate(connection, result, callback) { 
    connection.update(sql, [result], callback); 
}