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

Question

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

当我在下面插入 Wine 类别时，它工作正常，因为它没有使用异步函数。但是，当我使用 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() 在它们全部完成时运行？对于异步专家来说，答案可能是显而易见的......

Answer 1

使用 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.");
        });
}