0

我正在尝试使用 BulkWriteOperation 更新数百万个数据，但是当我的查询条件不满足但具有该 ID 的文档可用时，我的代码会给出异常。这是我的代码：-

if(provisionSubscriberList.size()>0){

        Map<String, Object> map = new HashMap<String, Object>();
        map.put("id", campaignTO.getId());
        map.put("testSample", false);
        map.put("status", "Active");
        map.put("controlGroup", false);
        try{
    WriteConcern wc = WriteConcern.ACKNOWLEDGED;
    BulkWriteOperation bulk = mongoTemplate.getCollection("provisionSubscriber").initializeOrderedBulkOperation();

    for (ProvisionSubscriberEntity provisionalSubscriber : provisionSubscriberList) {

        Query queryForAddSubscriber = new Query();

        Update updateFieldsForAddSubscriber = new Update();
        updateFieldsForAddSubscriber.set("msisdn", provisionalSubscriber.getMsisdn());
        updateFieldsForAddSubscriber.set("deviceType", provisionalSubscriber.getDeviceType());
        updateFieldsForAddSubscriber.addToSet("campaignIdList", map);


        List<DBObject> criteria = new ArrayList<DBObject>();
        criteria.add(new BasicDBObject("_id",new ObjectId(provisionalSubscriber.getId())));
        criteria.add(new BasicDBObject("campaignIdList.id", new BasicDBObject("$ne", campaignTO.getId())));
        criteria.add(new BasicDBObject("campaignIdList.controlGroup", new BasicDBObject("$ne", true)));
        criteria.add(new BasicDBObject("campaignIdList.status", new BasicDBObject("$ne", "Active")));
        BasicDBObject queryCriteria = new BasicDBObject("$and", criteria);

        bulk.find(queryCriteria).upsert().updateOne(updateFieldsForAddSubscriber.getUpdateObject());

    }
    BulkWriteResult results =bulk.execute(wc);
    System.out.println(results);
    for (BulkWriteUpsert up : results.getUpserts()) {
        System.out.println(up.getId());
    }

这是我得到的例外： -

com.mongodb.BulkWriteException: Bulk write operation error on server 192.168.1.113:27017. Write errors: [BulkWriteError{index=0, code=11000, message='E11000 duplicate key error index: jmailer_digiengage.provisionSubscriber.$_id_ dup key: { : ObjectId('58c8f33301de9614143f5812') }', details={ }}]. 
at com.mongodb.BulkWriteHelper.translateBulkWriteException(BulkWriteHelper.java:56)
at com.mongodb.DBCollection.executeBulkWriteOperation(DBCollection.java:2310)
at com.mongodb.BulkWriteOperation.execute(BulkWriteOperation.java:136)
at com.lumatadigital.digiengage.daoImpl.ProvisioningDaoImpl.provisionOnCampaign(ProvisioningDaoImpl.java:120)
at com.lumatadigital.digiengage.schedular.service.SchedularJobConfig.provisioningJob(SchedularJobConfig.java:29)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:606)
at org.springframework.util.MethodInvoker.invoke(MethodInvoker.java:269)
at org.springframework.scheduling.quartz.MethodInvokingJobDetailFactoryBean$MethodInvokingJob.executeInternal(MethodInvokingJobDetailFactoryBean.java:257)
at org.springframework.scheduling.quartz.QuartzJobBean.execute(QuartzJobBean.java:75)
at org.quartz.core.JobRunShell.run(JobRunShell.java:213)
at org.quartz.simpl.SimpleThreadPool$WorkerThread.run(SimpleThreadPool.java:557)

编辑：基本上，如果文档不可用，我想插入数据，或者如果文档可用并且我的查询对该文档满意，我想更新数据，否则跳过该文档。另外，我想跟踪插入的文档。

4

2 回答 2

1

这是因为在您的查询条件中：

List<DBObject> criteria = new ArrayList<DBObject>();
criteria.add(new BasicDBObject("_id",new ObjectId(provisionalSubscriber.getId())));
criteria.add(new BasicDBObject("campaignIdList.id", new BasicDBObject("$ne", campaignTO.getId())));
criteria.add(new BasicDBObject("campaignIdList.controlGroup", new BasicDBObject("$ne", true)));
criteria.add(new BasicDBObject("campaignIdList.status", new BasicDBObject("$ne", "Active")));
BasicDBObject queryCriteria = new BasicDBObject("$and", criteria);

如果 _id 字段已使用 insert 语句插入到数据库中，并且当 update 语句下一次运行时，活动列表对象中的条件“$ne”（不等于）将失败，这将创建带有相同的 _id 尝试插入而不是更新，因为以前的数据与当前数据不匹配。

因此，您收到以下错误：

E11000 duplicate key error index: jmailer_digiengage.provisionSubscriber.$_id_ dup key: { : ObjectId('58c8f33301de9614143f5812') }
于 2017-05-19T06:14:17.777 回答
0

要进行批量更新，您可以使用以下代码

        MongoCollection<Document> collection = database.getCollection("collection");
        List<WriteModel<Document>> updates = new ArrayList<WriteModel<Document>>();

        UpdateOptions options = new UpdateOptions();
        options.upsert(true);
        // Doc1 update
        Document doc1 = new Document("$set", new Document("key1", "value1"));
        updates.add(new UpdateOneModel<Document>(new Document("_id",new ObjectId("562a44971bca3c0001953f42")), doc1, options));

        //Doc2 update
        Document doc2 = new Document("$set", new Document("key1", "value2"));
        updates.add(new UpdateOneModel<Document>(new Document("_id",new ObjectId("562a44971bca3c0001954071")), doc2, options));

        BulkWriteResult result = collection.bulkWrite(updates);
        System.out.println("Updated count : " + result.getModifiedCount());

在下面的片段中

updates.add(new UpdateOneModel<Document>(new Document("_id",new ObjectId("562a44971bca3c0001954071")), doc2, options));

第一个条件是过滤条件，您可以使用文档中存在的任何键来过滤出要更新的文档，第二个参数是需要为文档更新的字段，第三个参数是可以使用的附加选项传递给模型

于 2017-05-19T06:48:05.810 回答