javascript - 从 mongodb nodejs 延迟返回集合数组

Question

我需要使用 express 和 mongodb 模块检索集合列表。
首先，我检索了一个有效的集合名称列表，然后我在循环中检索那些给定集合的数据。我的问题出在 getColAsync() 中:

getColAsync()   {
    return new Promise((resolve, reject)    =>    {
        this.connectDB().then((db)  =>  {
            var allCols = [];
            let dbase = db.db(this.databaseName);
            dbase.listCollections().toArray((err, collectionNames)  =>  {
                if(err) {
                    console.log(err);
                    reject(err);
                }
                else    {
                    for(let i = 0; i < collectionNames.length; i++) {
                        dbase.collection(collectionNames[i].name.toString()).find({}).toArray((err, collectionData) =>  {
                            console.log("current collection data: " + collectionData);
                            allCols[i] = collectionData;
                        })
                    }
                    console.log("done getting all data");
                    resolve(allCols);
                }
            })
        })
    })
}

connectDB() {
    if(this.dbConnection)   {
        // if connection exists
        return this.dbConnection;
    }
    else    {
        this.dbConnection = new Promise((resolve, reject) =>    {
            mongoClient.connect(this.URL, (err, db) =>  {
                if(err) {
                    console.log("DB Access: Error on mongoClient.connect.");
                    console.log(err);
                    reject(err);
                }
                else    {
                    console.log("DB Access: resolved.");
                    resolve(db);
                }
            });
        });
        console.log("DB Access: db exists. Connected.");
        return this.dbConnection;
    }
}

在我检索每个集合的 forloop 中，console.log("done getting all data") 被调用，promise 在 forloop 开始之前就被解决了。例如:

done getting all data
current collection data: something
current collection data: something2
current collection data: something3

请帮忙

Answer 1

问题

您的代码中的问题是这部分:

for (let i = 0; i < collectionNames.length; i++) {
    dbase.collection(collectionNames[i].name.toString()).find({}).toArray((err, collectionData) => {
        console.log("current collection data: " + collectionData);
        allCols[i] = collectionData;
    })
}
console.log("done getting all data");
resolve(allCols);

您应该注意到 resolve(allCols);在 for 之后立即调用循环结束，但循环的每次迭代都不会等待 toArray要调用的回调。

行 dbase.collection(collectionNames[i].name.toString()).find({}).toArray(callback)是异步的，因此循环将结束，您将调用 resolve(allCols); ，但 .find({}).toArray代码还没有完成。

解决方案概念

所以，基本上你所做的是:

初始化结果数组allCols = []

启动一系列异步操作

返回(仍然为空的)结果数组

随着异步操作完成，填充现在无用的结果数组。

你应该做的是:

启动一系列异步操作

等待全部 其中完成

从每一个中获取结果

返回结果列表

关键是 Promise.all([/* array of promises */])接受一组 promise 并返回 Promise 的函数本身向下游传递了一个包含所有结果的数组，所以我们需要获得的是这样的:

const dataPromises = []

for (let i = 0; i < collectionNames.length; i++) {
    dataPromises[i] = /* data fetch promise */;
}

return Promise.all(dataPromises);

如您所见，最后一行是 return Promise.all(dataPromises);而不是 resolve(allCols)和你的代码一样，所以我们不能再在 new Promise(func) 中执行此代码构造函数。

相反，我们应该链接 Promise与 .then()像这样:

getColAsync() {
    return this.connectDB().then((db) => {
        let dbase = db.db(this.databaseName);
        const dataPromises = []
        dbase.listCollections().toArray((err, collectionNames) => {
            if (err) {
                console.log(err);
                return Promise.reject(err);
            } else {
                for (let i = 0; i < collectionNames.length; i++) {
                    dataPromises[i] = new Promise((res, rej) => {
                        dbase.collection(collectionNames[i].name.toString()).find({}).toArray((err, collectionData) => {
                            console.log("current collection data: " + collectionData);
                            if (err) {
                                console.log(err);
                                reject(err);
                            } else {
                                resolve(collectionData);
                            }
                        });
                    });
                }
                console.log("done getting all data");
                return Promise.all(dataPromises);
            }
        });
    })
}

注意现在我们返回一个 return this.connectDB().then(...) ，然后返回 Promise.all(dataPromises);这个返回新的 Promise s 在每一步让我们保持 Promise链，因此 getColAsync()本身会返回 Promise然后你可以处理 .then()和 .catch() .

更清洁的代码

您可以稍微清理一下代码:

getColAsync() {
    return this.connectDB().then((db) => {
        let dbase = db.db(this.databaseName);
        const dataPromises = []
        // dbase.listCollections().toArray() returns a promise itself
        return dbase.listCollections().toArray()
            .then((collectionsInfo) => {
                // collectionsInfo.map converts an array of collection info into an array of selected
                // collections
                return collectionsInfo.map((info) => {
                    return dbase.collection(info.name);
                });
            })
    }).then((collections) => {
        // collections.map converts an array of selected collections into an array of Promises
        // to get each collection data.
        return Promise.all(collections.map((collection) => {
            return collection.find({}).toArray();
        }))
    })
}

如您所见，主要变化是:

以 promise 形式使用 mondodb 函数

使用 Array.map轻松将数据数组转换为新数组

下面我还将介绍您的代码的变体，它使用带有回调的函数和我正在处理的模块。

promise 混合

我最近在做 this npm module帮助获得更清晰、更易读的 Promises 组合。

在你的情况下，我会使用 fCombine处理您选择数据库并获取集合信息列表的第一步的函数:

Promise.fCombine({
    dbase: (dbURL, done) => mongoClient.connect(dbURL, done),
    collInfos: ({ dbase }, done) => getCollectionsInfo(dbase, done),
}, { dbURL: this.URL })

这会产生一个 promise ，向下游传递一个对象 {dbase: /* the db instance */, collInfos: [/* the list of collections info */]} .在哪里 getCollectionNames(dbase, done)是一个具有回调模式的函数，如下所示:

getCollectionsInfo = (db, done) => {
    let dbase = db.db(this.databaseName);
    dbase.listCollections().toArray(done);
}

现在您可以链接先前的 Promise 并将集合信息列表转换为选定的数据库集合，如下所示:

Promise.fCombine({
    dbase: ({ dbURL }, done) => mongoClient.connect(dbURL, done),
    collInfos: ({ dbase }, done) => getCollectionsInfo(dbase, done),
}, { dbURL: this.URL }).then(({ dbase, collInfos }) => {
    return Promise.resolve(collInfos.map((info) => {
        return dbase.collection(info.name);
    }));
})

现在下游我们有一个从数据库中选择的集合列表，我们应该从每个集合中获取数据，然后将结果与集合数据合并到一个数组中。
在我的模块中，我有一个 _mux创建 PromiseMux 的选项模仿常规 Promise 的行为和组成模式，但它实际上同时在处理多个 Promise。每个 Promise 从下游集合数组中输入一项，因此您可以编写代码以从通用集合中获取数据，它将为数组中的每个集合执行:

Promise.fCombine({
    dbase: ({ dbURL }, done) => mongoClient.connect(dbURL, done),
    collInfos: ({ dbase }, done) => getCollectionsInfo(dbase, done),
}, { dbURL: this.URL }).then(({ dbase, collInfos }) => {
    return Promise.resolve(collInfos.map((info) => {
        return dbase.collection(info.name);
    }));
})._mux((mux) => {
    return mux._fReduce([
        (collection, done) => collection.find({}).toArray(done)
    ]).deMux((allCollectionsData) => {
        return Promise.resolve(allCollectionsData);
    })
});

在上面的代码中， _fReduce行为类似于 _fCombine ，但它接受一个带有回调的函数数组而不是一个对象，并且它只向下游传递最后一个函数的结果(而不是具有所有结果的结构化对象)。最后 deMux执行 Promise.all在每个同时 Promise多路复用器，将他们的结果放在一起。

因此整个代码看起来像这样:

getCollectionsInfo = (db, done) => {
    let dbase = db.db(this.databaseName);
    dbase.listCollections().toArray(done);
}

getCollAsync = () => {
    return Promise.fCombine({
        /**
         * fCombine uses an object whose fields are functions with callback pattern to
         * build consecutive Promises. Each subsequent functions gets as input the results
         * from previous functions.
         * The second parameter of the fCombine is the initial value, which in our case is
         * the db url.
         */
        dbase: ({ dbURL }, done) => mongoClient.connect(dbURL, done), // connect to DB, return the connected dbase
        collInfos: ({ dbase }, done) => getCollectionsInfo(dbase, done), // fetch collection info from dbase, return the info objects
    }, { dbURL: this.URL }).then(({ dbase, collInfos }) => {
        return Promise.resolve(collInfos.map((info) => {
            /**
             *  we use Array.map to convert collection info into 
             *  a list of selected db collections
             */
            return dbase.collection(info.name);
        }));
    })._mux((mux) => {
        /**
         * _mux splits the list of collections returned before into a series of "simultaneous promises"
         * which you can manipulate as if they were a single Promise.
         */
        return mux._fReduce([ // this fReduce here gets as input a single collection from the retrieved list
            (collection, done) => collection.find({}).toArray(done)
        ]).deMux((allCollectionsData) => {
            // finally we can put back together all the results.
            return Promise.resolve(allCollectionsData);
        })
    });
}

在我的模块中，我试图避免最常见的反模式思想，我仍然会研究一些 Ghost Promise。

使用来自 mongodb 的 Promise 会变得更加简洁:

getCollAsync = () => {
    return Promise.combine({
        dbase: ({ dbURL }) => { return mongoClient.connect(dbURL); },
        collInfos: ({ dbase }) => {
            return dbase.db(this.databaseName)
                .listCollections().toArray();
        },
    }, { dbURL: this.URL }).then(({ dbase, collInfos }) => {
        return Promise.resolve(collInfos.map((info) => {
            return dbase.collection(info.name);
        }));
    }).then((collections) => {
        return Promise.all(collections.map((collection) => {
            return collection.find({}).toArray();
        }))
    });
}