续集等待直到循环完成回调

Question

来自 php 背景，我正试图了解这个回调的东西。

基本上我想得到一些行，然后我想遍历这些行并将它们与另一个模型（不同的数据库）进行检查。我希望回调等到它们都被循环并检查完毕。

在 sequelize 循环所有结果之前调用回调。

基本上我希望该功能是“阻塞”的。我需要改变什么？

toexport.getlasttransactions = function(lower,upper,callback){
    var deferred = Q.defer();
    var transactionsToUpdate = [];
    ///////////////////////////
    // set import conditions //
    ///////////////////////////
    var lowerbound = (lower) ? lower.format() : moment.utc().subtract(10, 'minutes').format();
    var upperbound = (upper) ? upper.format() : moment.utc().format();

    ///////////////////////////////
    // get IDs From Failed syncs //
    ///////////////////////////////
    FailedSync.find({ limit: 100 })
    .then(function(res){
        var FailedIDs = [];
        _.each(res, function(value,index){
            FailedIDs.push(value.transaction_id);
        });

        // build condition
        var queryCondition = { where: { updated_at: { between: [lowerbound,upperbound] } }, limit: 3 };
        if(FailedIDs.length > 0){
            queryCondition = {
                where: Sequelize.and({ updated_at: { between: [lowerbound,upperbound] } },
                Sequelize.or(
                  { id: FailedIDs }
                ))
            }
        }
        //////////////////////////////
        // get Phoenix Transactions //
        //////////////////////////////
        PhoenixTransaction
        .findAll(queryCondition)
        .then(function(poenixTrx){

            _.each(poenixTrx, function(value, index){

                Transaction.findOne({ where: { id: value.id }})
                .then(function(result){

                    if(!result || result.length === 0){
                        transactionsToUpdate.push(value);
                        console.log('!result || result.length === 0')
                    }
                    else if(result && result.length === 1){
                        if(result.hash != value.hash){
                            transactionsToUpdate.push(value);
                            console.log('result.hash != poenixTrx[i].hash')
                        }
                    }




                })
                .catch(function(err) {
                  console.log(err)
                })


            })
            deferred.resolve(transactionsToUpdate);


        })
        .catch(function(err){
          throw new Error("Something went wrong getting PhoenixTransaction") 
        })

    })

    deferred.promise.nodeify(callback);
    return deferred.promise;    

}

Answer 1

你的代码中有很多新承诺用户的模式：

• 您在不需要时使用延迟。
• 您没有使用 Promise 聚合方法
• 您不是在等待合适的地方，而是在嵌套。

Promise 代表一个随时间变化的价值。您可以稍后使用 Promise 并通过 then 访问其结果，而不仅仅是立即使用 - Sequelize 的 Promise 基于 bluebird，并提供了丰富的 API 为您进行聚合。 这是清理后代码的注释版本 - 请注意它不是嵌套：

toexport.getlasttransactions = function(lower,upper){ // no need for callback
    var lowerbound = (lower || moment.utc().subtract(10, 'minutes')).format();
    var upperbound = (upper || moment.utc()).format();
    // use `map` over a `each` with a push.
    var failedIds = FailedSync.find({ limit: 100 }).map(function(value){ 
        return value.transaction_id;
    });
    // build condition.
    var queryCondition = {
        where: { updated_at: { between: [lowerbound,upperbound] } }, limit: 3 
    };
    var query = failedIds.then(function(ids){ // use promise as proxy
        if(ids.length === 0) return queryCondition;
        return { // You can return a value or a promise from `then`
            where: Sequelize.and({ updated_at: { between: [lowerbound,upperbound] } },
                   Sequelize.or({ id: ids});
        };
    });
    var pheonixTransactions = query.then(function(condition){
        return PhoenixTransaction.findAll(queryCondition); // filter based on result
    });
    return pheonixTransactions.map(function(value){ // again, map over each
        return Transaction.findOne({ where: { id: value.id }}); // get the relevant one
    }).filter(function(result){ // filter over if chain and push
        return (!result || result.length === 0) || 
               ((result && result.length === 1) && result.hash != value.hash);
    });
};

Answer 2

理想情况下，您可能希望将 Bluebird 的 reduce 之类的东西与一系列 Promise 一起使用，但我将提供一个 async.series 实现，因为它更易于理解。

安装异步

npm install async

在你的文件中需要它

var async = require('async')

然后这样实现：

        //////////////////////////////
        // get Phoenix Transactions //
        //////////////////////////////
        PhoenixTransaction
        .findAll(queryCondition)
        .then(function(poenixTrx){

            var queryArray = poenixTrx.map(function(value){
                return function(callback){
                    Transaction.findOne({ where: { id: value.id }})
                    .then(function(result){

                        if(!result || result.length === 0){
                            transactionsToUpdate.push(value);
                            console.log('!result || result.length === 0')
                        }
                        else if(result && result.length === 1){
                            if(result.hash != value.hash){
                                transactionsToUpdate.push(value);
                                console.log('result.hash != poenixTrx[i].hash')
                            }
                        }

                        // trigger callback with any result you want
                        callback(null, result)
                    })
                    .catch(function(err) {
                      console.log(err)
                      // trigger  error callback
                      callback(err)
                    })
                }

            })

            // async.series will loop through he queryArray, and execute each function one by one until they are all completed or an error is thrown.
            // for additional information see https://github.com/caolan/async#seriestasks-callback
            async.series(queryArray, function(err, callback){
                // after all your queries are done, execution will be here
                // resolve the promise with the transactionToUpdate array
                deferred.resolve(transactionsToUpdate);
            })


        })
        .catch(function(err){
          throw new Error("Something went wrong getting PhoenixTransaction") 
        })

Answer 3

说实话，整件事有点乱。尤其是承诺/回调混淆可能会在某些时候给你带来问题。无论如何，您在 transactionsToUpdate 上使用 deferred.resolve 这只是一个数组，因此它会立即调用回调。

如果您保留该脚本，而不是使用 _.each 之类的异步 (https://github.com/caolan/async) 来并行运行您的事务并将其用作回调。

它可能看起来像这样：

toexport.getlasttransactions = function(lower,upper,callback){
    var transactionsToUpdate = [];
    ///////////////////////////
    // set import conditions //
    ///////////////////////////
    var lowerbound = (lower) ? lower.format() : moment.utc().subtract(10, 'minutes').format();
    var upperbound = (upper) ? upper.format() : moment.utc().format();

    ///////////////////////////////
    // get IDs From Failed syncs //
    ///////////////////////////////
    FailedSync.find({ limit: 100 })
    .then(function(res){
        var FailedIDs = [];
        _.each(res, function(value,index){
            FailedIDs.push(value.transaction_id);
        });

        // build condition
        var queryCondition = { where: { updated_at: { between: [lowerbound,upperbound] } }, limit: 3 };
        if(FailedIDs.length > 0){
            queryCondition = {
                where: Sequelize.and({ updated_at: { between: [lowerbound,upperbound] } },
                Sequelize.or(
                  { id: FailedIDs }
                ))
            }
        }
        //////////////////////////////
        // get Phoenix Transactions //
        //////////////////////////////
        PhoenixTransaction
        .findAll(queryCondition)
        .then(function(poenixTrx){

            async.each(poenixTrx, function(value, next){

                Transaction.findOne({ where: { id: value.id }})
                .then(function(result){

                    if(!result || result.length === 0){
                        transactionsToUpdate.push(value);
                        console.log('!result || result.length === 0')
                    }
                    else if(result && result.length === 1){
                        if(result.hash != value.hash){
                            transactionsToUpdate.push(value);
                            console.log('result.hash != poenixTrx[i].hash')
                        }
                    }

                    next();
                })
                .catch(function(err) {
                  console.log(err)
                })


            }, function(err) {
              //Return the array transactionsToUpdate in your callback for further use
              return callback(err, transactionsToUpdate);
            });
        })
        .catch(function(err){
          throw new Error("Something went wrong getting PhoenixTransaction") 
        })

    })

}

这将是回调的方式。 但是您需要确定要使用的内容：回调或承诺。不要同时使用两者（例如：如果您的方法需要回调，则不应返回 Promise，或者如果返回 Promise，则不应预期回调）。

如果你使用回调，你不想抛出错误，你只需调用回调并在回调中给出错误 - 使用你的方法的人可以从回调中检查错误并处理它。

希望这对你有点意义，我知道如果你来自 php 之类的东西，那么整个回调和承诺有点奇怪，它需要一些习惯:)

Answer 4

感谢您解释差异。我认为使用 Promise 是前进的方向，因为它使代码看起来更好，并避免了这种“回调地狱”。

例如：

PhoenixSyncTransactions.getlasttransactions(lastTimeSynced,null)
.then(function(res){

    return PersistTransaction.prepareTransactions(res).then(function(preparedTrx){
      return preparedTrx;
    })
}).then(function(preparedTrx){
    return PersistTransaction.persistToDB(preparedTrx).then(function(Processes){
      return Processes;
    })
})
.then(function(Processes){
    return PersistTransaction.checkIfMultiProcess(Processes).then(function(result){
      return result;
    })

})
.then(function(result){
  console.log('All jobs done');
})

整个代码更容易阅读。