节点&MySQL:Can't端连接->异步混淆

Node & MySQL: Can't end connection -> Async confusion

本文关键字：连接 gt 异步 MySQL amp Can 节点      更新时间：2023-09-26

我正试图编写一个Node程序，用磁盘上文件中的数据填充MySQL数据库。我可能会也可能不会以正确的方式来做这件事，但它正在发挥作用。我遇到的问题是理解我应该如何处理允许异步函数在与DB的连接结束之前完成的问题。最终，我将读取大量的数据文件，并像下面那样将它们插入数据库。我可以只使用readFileSync而不是异步版本，但我需要更好地处理异步函数。

当我在下面插入葡萄酒类别时，它工作得很好，因为它没有使用异步函数。然而，当我使用readFile从文件中获取数据时，我会得到一个错误，即在执行任何查询之前连接就结束了：

connection.connect( function(err) {
    if(err) {
        console.log(err);
    }
});
// Take a table and the values, and insert a new row into a table
function insert_into( table, values ) {
    if( values instanceof Array ) {
        values = values.map( function( value ) {
            return '"' + value + '"';
        }).join(', ');
    } else {
        values = '"' + values + '"';
    }
    var statement = 'INSERT INTO ' + table + ' VALUES (NULL, ' + values + ')';
    connection.query( statement, function(err, rows, fields) {
      if (err) throw err;
      console.log( values + " successfully added.");
    });
};
// Populate the wine_categories table
var wine_categories = [
    'red', 'white', 'rose', 'sparkling', 'fortified'
];
// Works fine when used alone
wine_categories.forEach( function( element ) {
    insert_into( 'wine_categories', element );
});
// Populate the countries table
// connection.end() runs before this finishes its job
fs.readFile( countries, 'utf8', function (err, data) {
    if (err) {
        throw err;
    } else {
        var codes = Array.prototype.map.call( 
            data.split(''n'), function( country ) {
                return country.split(''t');
        });
        codes.forEach( function( country ) {
            if( country[1].length > 25 ) {
                country[1] = country[1].substring(0, 25);
            }
            insert_into( 'countries', country );
        });
    }
}); 
connection.end();

显然，connection.end()需要在所有插入完成之后发生，但我不确定如何处理。我不希望它是readFile调用的回调，因为我最终会在这个文件中有许多类似的调用。

我应该如何构建我的代码，以便在所有查询完成后执行并运行connection.end()？答案对于异步向导来说可能是显而易见的。。。

如果使用promise，它将是这样的：

pool.getConnectionAsync().then(function(connection) {
    // Populate the wine_categories table
    var wine_categories = [
        'red', 'white', 'rose', 'sparkling', 'fortified'
    ];
    var wineQueries = wine_categories.map(function(wine){
        return insert_into(connection, "wine_categories", wine);
    });
    var countryQueries = fs.readFileAsync(countries, "utf-8").then(function(data) {
        return data.split("'n").map(function(country) {
            country = country.split("'t")[1];
            if (country.length > 25) {
                country = country.substring(0, 25);
            }
            return insert_into(connection, "countries", country);
        });
    });
    Promise.all(wineQueries.concat(countryQueries))
        .then(function() {
            console.log("all done");
        })
        .catch(function(e) {
            console.log("error", e);
        })
        .finally(function() {
            connection.release();
        })
});

上述的先决条件代码

var Promise = require("bluebird");
var fs = Promise.promisifyAll(require("fs"));
Promise.promisifyAll(require("mysql/lib/Connection").prototype);
var pool = Promise.promisifyAll(require("mysql").createPool({
    "user": "...",
    "password": "...",
    "database": "...",
    "host": "localhost",
    "port": 3306,
    "debug": false
}));
function insert_into(connection, table, values) {
    if( values instanceof Array ) {
        values = values.map(connection.escape, connection).join(', ');
    } else {
        values = connection.escape(values);
    }
    return connection
        .queryAsync('INSERT INTO ' + table + ' VALUES (NULL, ' + values + ')')
        .then(function() {
            console.log(values + " successfully added.");
        });
}

假设insert_into也是异步的，您可能希望使用类似async.each的东西来处理插入记录。它有一个方便的回调，当插入所有记录时会调用它，因为只有在那一点上你才想关闭连接：

async.each(codes, function(country, callback) {
  if ( country[1].length > 25 ) {
    country[1] = country[1].substring(0, 25);
  }
  insert_into( 'countries', country, callback ); // !! read below
}, function(err) {
  // TODO: handle any errors
  ...
  // Here, all countries are inserted.
  connection.end();
});

然而，这意味着insert_into也应该接受回调（使用公共节点约定function(err, result)），该回调将在插入记录时调用。在上面的代码中，我直接使用async提供的回调，这意味着一旦您的insert_into完成，它将调用async回调信号，表明each的迭代已经完成。

EDIT：您可以重写insert_into，使其看起来像这样：

function insert_into( table, values, callback ) {
  ...
  connection.query(..., function(err) {
    callback(err);
  });
}

由于您不需要connection.query的实际结果，所以您只需要传递err（而不是抛出它）。

提示：假设您使用的是node-mysql，您可能需要查看文档，了解它如何帮助您进行转义。

相关文章：