Async.waterfall：callback已经被调用？

我知道这是以前在这里提到的一个话题，但我使用rethinkdb async.waterfall，我得到Error: Callback was already called 。 奇怪的是，即使它抛出的错误和崩溃的应用程序，它仍然会创build我所需要的数据库和表。 我已经阅读了几个像NodeJS Async：Callback已经被调用的答案吗？ 或使用async.waterfall ，但我似乎无法得到任何地方。 我的控制台也告诉我，错误是在db.js:40:9 ，但我是新来的节点，只是不知道它想要什么callback。 我究竟做错了什么？ 我需要在这里嵌套callback吗？ 我正在使用的代码如下所示。 任何帮助，我可以在这里非常感激，如果需要我可以张贴其他相关的代码。 多谢你们。

db.js：

  exports.setDatabaseAndTables = function() { async.waterfall([ function connect(callback) { r.connect(config.rethinkdb, callback); }, function createDatabase(connection, callback) { // Create the database if needed. r.dbList().contains(config.rethinkdb.db).do(function(containsDb) { return r.branch( containsDb, {created: 0}, r.dbCreate(config.rethinkdb.db) ); }).run(connection, function(err) { callback(err, connection); }); }, function createTable(connection, callback) { // Create the tables if needed. r.tableList().contains('todos').do(function(containsTable) { return r.branch( containsTable, {created: 0}, r.tableCreate('todos') ); }).run(connection, function(err) { callback(err, connection); }); r.tableList().contains('users').do(function(containsTable) { return r.branch( containsTable, {created: 0}, r.tableCreate('users') ); }).run(connection, function(err) { callback(err, connection); }); }, function createIndex(connection, callback) { // Create the indexes if needed. r.table('todos').indexList().contains('createdAt').do(function(hasIndex) { return r.branch( hasIndex, {created: 0}, r.table('todos').indexCreate('createdAt') ); }).run(connection, function(err) { callback(err, connection); }); r.table('users').indexList().contains('createdAt').do(function(hasIndex) { return r.branch( hasIndex, {created: 0}, r.table('users').indexCreate('createdAt') ); }).run(connection, function(err) { callback(err, connection); }); }, function waitForIndex(connection, callback) { // Wait for the index to be ready. r.table('todos').indexWait('createdAt').run(connection, function(err, result) { callback(err, connection); }); r.table('users').indexWait('createdAt').run(connection, function(err, result) { callback(err, connection); }); } ], function(err, connection) { if(err) { console.error(err); process.exit(1); return; } }); }; 

Answer 1

对于那些尝试“asynchronousstream水”的人来说，这是一个普遍的问题。

下面是通过将“createTable”，“createIndex”和“waitForIndex”分别分成两个函数的解决scheme：

 exports.setDatabaseAndTables = function() { async.waterfall([ function connect(callback) { r.connect(config.rethinkdb, callback); }, function createDatabase(connection, callback) { // Create the database if needed. r.dbList().contains(config.rethinkdb.db).do(function(containsDb) { return r.branch( containsDb, {created: 0}, r.dbCreate(config.rethinkdb.db) ); }).run(connection, function(err) { callback(err, connection); }); }, function createTableTodos(connection, callback) { // Create the tables if needed. r.tableList().contains('todos').do(function(containsTable) { return r.branch( containsTable, {created: 0}, r.tableCreate('todos') ); }).run(connection, function(err) { callback(err, connection); }); }, function createTableUsers(connection, callback) { r.tableList().contains('users').do(function(containsTable) { return r.branch( containsTable, {created: 0}, r.tableCreate('users') ); }).run(connection, function(err) { callback(err, connection); }); }, function createIndexTodos(connection, callback) { // Create the indexes if needed. r.table('todos').indexList().contains('createdAt').do(function(hasIndex) { return r.branch( hasIndex, {created: 0}, r.table('todos').indexCreate('createdAt') ); }).run(connection, function(err) { callback(err, connection); }); }, function createIndexUsers(connection, callback) { r.table('users').indexList().contains('createdAt').do(function(hasIndex) { return r.branch( hasIndex, {created: 0}, r.table('users').indexCreate('createdAt') ); }).run(connection, function(err) { callback(err, connection); }); }, function waitForIndexTodos(connection, callback) { // Wait for the index to be ready. r.table('todos').indexWait('createdAt').run(connection, function(err, result) { callback(err, connection); }); }, function waitForIndexUsers(connection, callback) { r.table('users').indexWait('createdAt').run(connection, function(err, result) { callback(err, connection); }); } ], function(err, connection) { if(err) { console.error(err); process.exit(1); return; } }); };