有没有在mongoose中执行批量upsert的选项？基本上是有一个数组，如果元素不存在，则插入该元素，如果存在，则更新它（我正在使用自定义_ids）
当我使用.insert时，MongoDB返回重复键错误E11000（应该更新），但是插入多个新文档确实可以正常工作：
var Users = self.db.collection('Users');
Users.insert(data, function(err){
if (err) {
callback(err);
}
else {
callback(null);
}
});
使用.save会返回一个错误，指出参数必须是单个文档:
Users.save(data, function(err){
...
}
谢谢！
.collection 访问器，该访问器基本上访问了 "集合对象"，而该对象是由 mongoose 本身实现的 "node native driver" 的基础。 var mongoose = require('mongoose'),
Schema = mongoose.Schema;
mongoose.connect('mongodb://localhost/test');
var sampleSchema = new Schema({},{ "strict": false });
var Sample = mongoose.model( "Sample", sampleSchema, "sample" );
mongoose.connection.on("open", function(err,conn) {
var bulk = Sample.collection.initializeOrderedBulkOp();
var counter = 0;
// representing a long loop
for ( var x = 0; x < 100000; x++ ) {
bulk.find(/* some search */).upsert().updateOne(
/* update conditions */
});
counter++;
if ( counter % 1000 == 0 )
bulk.execute(function(err,result) {
bulk = Sample.collection.initializeOrderedBulkOp();
});
}
if ( counter % 1000 != 0 )
bulk.execute(function(err,result) {
// maybe do something with result
});
});
.update()默认只修改找到的“第一个”文档。喜欢这种方法吗？看看那些一无所知的评论者和那些真正拥有知识的回答者之间的巨大差异？ - Neil Lunn.execute()之后，“批量”操作才是异步的。这是因为向服务器来回传输数据会花费大量的IO，所以您要尽量减少此类操作。虽然在同步循环中可能会多次使用.execute()并使用多个连接，但您可以通过类似async.whilst或其他控件来改变它，其中迭代可以由回调来控制（因此在.execute()内部）以处理完成。这在Promises中有些困难，但仍然是可能的。 - Neil Lunnulimit 的帮助下发现，如果 Node 创建文档的速度比 Mongo 存储它们（以及网络传输它们）的速度快，那么两者都可以填满 Node 的内存。因此，对于大量文档，我们应该限制一次执行的批量操作数量，通过等待完成来实现。同时运行 4 个并行操作可能是一个不错的方法。 - joeytwiddle不需要像@neil-lunn建议的那样管理限制(1000)。 Mongoose已经处理了这个问题。 我以他的优秀答案为基础，完成了这个完整的基于Promise的实现和示例：
var Promise = require('bluebird');
var mongoose = require('mongoose');
var Show = mongoose.model('Show', {
"id": Number,
"title": String,
"provider": {'type':String, 'default':'eztv'}
});
/**
* Atomic connect Promise - not sure if I need this, might be in mongoose already..
* @return {Priomise}
*/
function connect(uri, options){
return new Promise(function(resolve, reject){
mongoose.connect(uri, options, function(err){
if (err) return reject(err);
resolve(mongoose.connection);
});
});
}
/**
* Bulk-upsert an array of records
* @param {Array} records List of records to update
* @param {Model} Model Mongoose model to update
* @param {Object} match Database field to match
* @return {Promise} always resolves a BulkWriteResult
*/
function save(records, Model, match){
match = match || 'id';
return new Promise(function(resolve, reject){
var bulk = Model.collection.initializeUnorderedBulkOp();
records.forEach(function(record){
var query = {};
query[match] = record[match];
bulk.find(query).upsert().updateOne( record );
});
bulk.execute(function(err, bulkres){
if (err) return reject(err);
resolve(bulkres);
});
});
}
/**
* Map function for EZTV-to-Show
* @param {Object} show EZTV show
* @return {Object} Mongoose Show object
*/
function mapEZ(show){
return {
title: show.title,
id: Number(show.id),
provider: 'eztv'
};
}
// if you are not using EZTV, put shows in here
var shows = []; // giant array of {id: X, title: "X"}
// var eztv = require('eztv');
// eztv.getShows({}, function(err, shows){
// if(err) return console.log('EZ Error:', err);
// var shows = shows.map(mapEZ);
console.log('found', shows.length, 'shows.');
connect('mongodb://localhost/tv', {}).then(function(db){
save(shows, Show).then(function(bulkRes){
console.log('Bulk complete.', bulkRes);
db.close();
}, function(err){
console.log('Bulk Error:', err);
db.close();
});
}, function(err){
console.log('DB Error:', err);
});
// });
这样做的好处是在完成后关闭连接，如果你关心的话，显示任何错误，但如果不关心则忽略它们（Promises中的错误回调是可选的）。而且它也非常快。我在这里分享我的发现。如果您想将所有eztv节目保存到数据库中，可以取消注释eztv相关内容，作为示例。
bulk.execute 的作用。https://docs.mongodb.org/v3.0/reference/method/db.collection.initializeUnorderedBulkOp/ - konsumerawait Model.bulkWrite(docs.map(doc => ({
updateOne: {
filter: {id: doc.id},
update: doc,
upsert: true
}
})))
或者更加详细:
const bulkOps = docs.map(doc => ({
updateOne: {
filter: {id: doc.id},
update: doc,
upsert: true
}
}))
Model.bulkWrite(bulkOps)
.then(bulkWriteOpResult => console.log('BULK update OK:', bulkWriteOpResult))
.catch(err => console.error('BULK update error:', err))
upsertMany方法，以执行批量upsert操作并提供了Promise接口。如果您在db.collection中没有看到bulk方法，即出现错误如xxx变量没有方法：initializeOrderedBulkOp（）
尝试更新您的mongoose版本。 显然，旧的mongoose版本不能传递所有基础的mongo db.collection方法。
npm install mongoose
对我来说解决了这个问题。
你可以使用mongoose的Model.bulkWrite()
const res = await Character.bulkWrite([
{
updateOne: {
filter: { name: 'Will Riker' },
update: { age: 29 },
upsert: true
}
},
{
updateOne: {
filter: { name: 'Geordi La Forge' },
update: { age: 29 },
upsert: true
}
}
]);
最近我在我的电子商务应用程序中存储产品时不得不实现这一点。由于我每4个小时必须插入10000个项目，所以我的数据库经常超时。对我来说，一种选择是在连接到数据库时在mongoose中设置socketTimeoutMS和connectTimeoutMS，但这种方法感觉有些hacky，而且我不想操纵数据库的连接超时默认值。我还看到@neil lunn的解决方案采用了一个简单的同步方法，在for循环内部进行取模运算。这里是我的异步版本，我相信它可以更好地完成工作。
let BATCH_SIZE = 500
Array.prototype.chunk = function (groupsize) {
var sets = [];
var chunks = this.length / groupsize;
for (var i = 0, j = 0; i < chunks; i++ , j += groupsize) {
sets[i] = this.slice(j, j + groupsize);
}
return sets;
}
function upsertDiscountedProducts(products) {
//Take the input array of products and divide it into chunks of BATCH_SIZE
let chunks = products.chunk(BATCH_SIZE), current = 0
console.log('Number of chunks ', chunks.length)
let bulk = models.Product.collection.initializeUnorderedBulkOp();
//Get the current time as timestamp
let timestamp = new Date(),
//Keep track of the number of items being looped
pendingCount = 0,
inserted = 0,
upserted = 0,
matched = 0,
modified = 0,
removed = 0,
//If atleast one upsert was performed
upsertHappened = false;
//Call the load function to get started
load()
function load() {
//If we have a chunk to process
if (current < chunks.length) {
console.log('Current value ', current)
for (let i = 0; i < chunks[current].length; i++) {
//For each item set the updated timestamp to the current time
let item = chunks[current][i]
//Set the updated timestamp on each item
item.updatedAt = timestamp;
bulk.find({ _id: item._id })
.upsert()
.updateOne({
"$set": item,
//If the item is being newly inserted, set a created timestamp on it
"$setOnInsert": {
"createdAt": timestamp
}
})
}
//Execute the bulk operation for the current chunk
bulk.execute((error, result) => {
if (error) {
console.error('Error while inserting products' + JSON.stringify(error))
next()
}
else {
//Atleast one upsert has happened
upsertHappened = true;
inserted += result.nInserted
upserted += result.nUpserted
matched += result.nMatched
modified += result.nModified
removed += result.nRemoved
//Move to the next chunk
next()
}
})
}
else {
console.log("Calling finish")
finish()
}
}
function next() {
current++;
//Reassign bulk to a new object and call load once again on the new object after incrementing chunk
bulk = models.Product.collection.initializeUnorderedBulkOp();
setTimeout(load, 0)
}
function finish() {
console.log('Inserted ', inserted + ' Upserted ', upserted, ' Matched ', matched, ' Modified ', modified, ' Removed ', removed)
//If atleast one chunk was inserted, remove all items with a 0% discount or not updated in the latest upsert
if (upsertHappened) {
console.log("Calling remove")
remove()
}
}
/**
* Remove all the items that were not updated in the recent upsert or those items with a discount of 0
*/
function remove() {
models.Product.remove(
{
"$or":
[{
"updatedAt": { "$lt": timestamp }
},
{
"discount": { "$eq": 0 }
}]
}, (error, obj) => {
if (error) {
console.log('Error while removing', JSON.stringify(error))
}
else {
if (obj.result.n === 0) {
console.log('Nothing was removed')
} else {
console.log('Removed ' + obj.result.n + ' documents')
}
}
}
)
}
}