node.js - nodejs和mongoskin，保存所有项目后的回调

Question

我有以下代码片段，我在其中迭代集合并执行另一个数据库查询并在其回调中构造一个对象。最后我将该对象保存到另一个集合中。

我希望在保存所有项目后调用另一个函数，但不知道如何调用。我尝试使用异步库，特别是当 item 不为空时使用异步库，但这只会让我陷入无限循环。

有没有办法确定所有项目何时已保存？

谢谢!

var cursor = db.collection('user_apps').find({}, {timeout:false});
cursor.each(function (err, item) {
    if (err) {
        throw err;
    }
    if (item) {
        var appList = item.appList;
        var uuid= item.uuid;
        db.collection('app_categories').find({schema_name:{$in: appList}}).toArray(function (err, result) {
            if (err) throw err;
            var catCount = _.countBy(result, function (obj) {
                return obj.category;
            })
            catObj['_id'] = uuid;
            catObj['total_app_num'] = result.length;
            catObj['app_breakdown'] = catCount;
            db.collection('audiences').insert(catObj, function (err) {
                if (err) console.log(err);
            });
        }); 
    }
    else {
        // do something here after all items have been saved
    }
});

Answer 1

这里的关键是使用在执行“循环”操作时尊重回调信号的东西。此处实现的 .each() 不会执行此操作，因此您需要一个“异步”循环控制，该控制将表示每个循环已迭代并完成，并在回调中具有自己的回调。

如果您的底层 MongoDB 驱动程序至少是版本 2，那么就会有 .forEach()它有一个回调，当循环完成时被调用。这比 .each() 更好，但它没有解决了解内部“异步”.insert() 操作何时完成的问题。

所以更好的方法是使用 stream interface由 .find() 返回，其中允许更多的流量控制。有一个 .stream() 方法用于向后兼容，但现代驱动程序默认只返回接口(interface):

var stream = db.collection('user_apps').find({});

stream.on("err",function(err){
    throw(err);
});

stream.on("data",function(item) {
    stream.pause();                 // pause processing of stream
    var appList = item.appList;
    var uuid= item.uuid;
    db.collection('app_categories').find({schema_name:{ "$in": appList}}).toArray(function (err, result) {
        if (err) throw err;
        var catCount = _.countBy(result, function (obj) {
            return obj.category;
        })

        var catObj = {};        // always re-init
        catObj['_id'] = uuid;
        catObj['total_app_num'] = result.length;
        catObj['app_breakdown'] = catCount;
        db.collection('audiences').insert(catObj, function (err) {
            if (err) console.log(err);
            stream.resume();        // resume stream processing
        });
    }); 
});

stream.on("end",function(){
    // stream complete and processing done
});

流上的 .pause() 方法会停止发出更多事件，以便一次处理一个对象结果。当调用 .insert() 的回调时，就会调用 .resume() 方法，表示该项目的处理已完成，可以进行新的调用来处理下一个项目。

当流完成时，一切都已完成，因此将调用“end”事件 Hook 来继续您的代码。

这样，每个循环都以结束来表示移动到下一次迭代，并且有一个定义的“结束”事件来完成处理的结束。由于控件位于 .insert() 回调“内部”，因此这些操作也将完成。

作为旁注，您可能会考虑在源集合中包含“类别”信息，因为使用 .aggregate() 似乎可以更有效地返回结果。如果所有必需的数据都在一个集合中。