Node 和 MySQL：无法结束连接 -> 异步混淆

Posted 2023-02-24

【中文标题】Node 和 MySQL：无法结束连接 -> 异步混淆

【英文标题】：Node & MySQL: Can't end connection -> Async confusion

【发布时间】：2014-01-01 23:01:07

【问题描述】：

我正在尝试编写一个 Node 程序，用我在磁盘上的文件中的数据填充我的 mysql 数据库。我可能会或可能不会以正确的方式进行此操作，但它正在起作用。我遇到的问题是了解我应该如何处理允许异步函数在与数据库的连接结束之前完成。最终，我将读取大量数据文件，并将它们插入到数据库中，就像我在下面所做的那样。我可以只使用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() 在它们全部完成后运行？对于异步向导来说，答案可能很明显......

【参考方案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.");
        );

【讨论】：

谢谢！对country 格式稍作修改后，它就可以工作了。几个后续问题： 1. 如果我有更多我想使用的插入查询，我可以继续将.concat(queryName) 添加到Promise.all 吗？ 2. 如果其他表具有正在填充的表的外键，我是否需要创建一个将在此之后运行的新池？

@EmptyArsenal Promise.all 接受一个承诺数组并返回一个承诺，当原始数组中的所有承诺都被履行时，该承诺将被履行。要运行查询，请从池中获取连接，然后使用该连接运行查询，最后释放连接。像这样：pastebin.com/JhyvFhAS

@EmptyArsenal 你可以创建一个这样的辅助函数来使它更容易pastebin.com/5twg6YE1

您在许多函数的末尾都有Async。这是 Bluebird 的约定，还是 readFileAsync 是 readFile 的同义词？

@EmptyArsenal promisifyAll 在带有Async 后缀的对象上创建方法。因此，例如 readFile 变为 readFileAsync。在 fs 和 mysql 等大多数库中，通常没有承诺返回方法。后缀用于避免覆盖。

【参考方案2】：

假设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的这次迭代已经完成。

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

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

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

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

【讨论】：

感谢您的指导。我在代码中添加了insert_into 函数。我看到你的建议看起来更像节点，我仍然在掌握。那么，在这种情况下，我可以只记录成功或其他内容作为回调，这就足够了？

@EmptyArsenal 查看我的编辑以获取有关如何重写 insert_into 以使其适合 async 工作流程的建议