/ /

批量写入操作

Overview

在本指南中，您可以学习如何在Java驱动程序中使用批量操作。

要执行单个创建、替换、更新或删除操作，可以使用相应的方法。示例，要插入一个文档并替换一个文档，可以使用 insertOne() 和 replaceOne() 方法。使用这些方法时，客户端会对每项操作调用数据库一次。

通过使用批量写入操作，您可以通过更少的数据库调用来执行多个写入操作。您可以在以下级别执行批量写入操作：

集合：您可以使用MongoCollection.bulkWrite() 方法对单个集合执行批量写入。在此方法中，每种写入操作都需要至少一次数据库调用。示例，MongoCollection.bulkWrite() 将多个更新操作放在一次调用中，但对数据库进行两次单独的调用以执行插入操作和替换操作。
客户端：如果您的应用程序连接到MongoDB Server8.0 或更高版本，则可以使用MongoClient.bulkWrite() 方法对同一集群中的多个集合和数据库执行批量写入操作。此方法在一次数据库调用中执行所有写入。

集合批量写入

批量写入操作包含一个或多个写入操作。要在集合级别执行批量写入操作，请将 List 的 WriteModel 文档传递给 MongoCollection.bulkWrite() 方法。 WriteModel 是表示写入操作的模型。

MongoCollection.bulkWrite() 方法在单独的数据库调用中执行每种写入。示例，当您将 DeleteOneModel、DeleteManyModel 和 ReplaceOneModel 对象传递给该方法时，该方法会执行两次调用：一次针对删除操作，另一次针对替换操作。

注意

当客户端将操作拆分为单独的数据库调用时，如果批量写入操作未进行排序，它可能会对操作进行重新排序以提高效率。要学习；了解有关操作执行顺序的更多信息，请参阅执行顺序部分。

以下各节介绍如何创建和使用每个 WriteModel 文档。每节中的示例都使用 people 集合中的以下文档：

{ "_id": 1, "name": "Karen Sandoval", "age": 31 }
{ "_id": 2, "name": "William Chin", "age": 54 }
{ "_id": 8, "name": "Shayla Ray", "age": 20 }

有关本节中提到的方法和类的详情，请参阅以下 API 文档：

插入操作

如需执行插入操作，请创建 InsertOneModel，指定要插入的文档。如需插入多个文档，必须为每个要插入的文档创建一个 InsertOneModel。

例子

以下示例为两个描述人物的文档创建了 InsertOneModel：

InsertOneModel<Document> juneDoc = new InsertOneModel<>(new Document("name", "June Carrie")
                                                            .append("age", 17));
                                                  
InsertOneModel<Document> kevinDoc = new InsertOneModel<>(new Document("name", "Kevin Moss")
                                                            .append("age", 22));

重要

执行 bulkWrite() 时，InsertOneModel 不能插入集合已存在的具有 _id 的文档。相反，该方法会抛出 MongoBulkWriteException。

以下示例尝试插入两个文档，其中 _id 是 1 和 3：

try {
    List<WriteModel<Document>> bulkOperations = new ArrayList<>();
    // Creates instructions to insert documents
    InsertOneModel<Document> doc1 = new InsertOneModel<>(new Document("_id", 1));
    InsertOneModel<Document> doc3 = new InsertOneModel<>(new Document("_id", 3));
    
    bulkOperations.add(doc1);
    bulkOperations.add(doc3);
    
    // Runs a bulk write operation for the specified insert WriteModels
    collection.bulkWrite(bulkOperations); 
// Prints a message if any exceptions occur during the bulk write operation
} catch (MongoBulkWriteException e){
    System.out.println("A MongoBulkWriteException occurred with the following message: " + e.getMessage());
}

以下显示了上述代码的输出：

A MongoBulkWriteException occurred with the following message:
Bulk write operation error on server sample-shard-00-02.pw0q4.mongodb.net:27017.
Write errors: [BulkWriteError{index=0, code=11000, message='E11000 duplicate key
error collection: crudOps.bulkWrite index: _id_ dup key: { _id: 1 }', details={}}].

要了解为何没有插入 _id 为 3 的文档，请参阅执行顺序部分。

有关本节提及的方法和类的详情，请参阅 InsertOneModel API 文档。

替换操作

要执行替换操作，请创建 ReplaceOneModel，为要替换为替换文档的文档指定查询筛选器。

重要

执行 bulkWrite() 时，ReplaceOneModel 无法对违反集合唯一索引约束的文档进行更改，并且如果查询筛选器没有匹配项，则模型不会替换文档。

例子

以下示例创建 ReplaceOneModel，用包含新增 location 字段的文档替换 _id 为 1 的文档：

ReplaceOneModel<Document> celineDoc = new ReplaceOneModel<>(
                                    Filters.eq("_id", 1), 
                                    new Document("name", "Celine Stork")
                                        .append("location", "San Diego, CA"));

如果多个文档与 ReplaceOneModel实例中指定的查询过滤匹配，则该操作将替换第一个结果。您可以在 ReplaceOptions实例中指定排序，以便在服务器执行替换操作之前对匹配的文档应用顺序，如以下代码所示：

ReplaceOptions options = ReplaceOptions.sort(Sorts.ascending("_id"));

有关部分提及的方法和类的更多信息，请参阅以下资源：

ReplaceOneModel API 文档
ReplaceOptions API文档
唯一索引服务器手册说明

更新操作

要执行更新操作，请创建 UpdateOneModel 或 UpdateManyModel 从而为要更新的文档指定查询筛选器。

UpdateOneModel 更新与查询筛选器匹配的第一个文档，UpdateManyModel 更新与查询筛选器匹配的所有文档。

重要

执行 bulkWrite() 时，UpdateOneModel 和 UpdateManyModel 无法对违反集合唯一索引约束的文档进行更改，并且如果查询筛选器没有匹配项，则模型不会更新任何文档。

例子

以下示例创建一个 UpdateOneModel 来更新文档中的 age 字段，其中 _id 为 2：

UpdateOneModel<Document> updateDoc = new UpdateOneModel<>(
                                    Filters.eq("_id", 2), 
                                    Updates.set("age", 31));

如果多个文档与 UpdateOneModel实例中指定的查询过滤匹配，则该操作会更新第一个结果。您可以在 UpdateOptions实例中指定排序，以便在服务器执行更新操作之前对匹配的文档应用顺序，如以下代码所示：

UpdateOptions options = UpdateOptions.sort(Sorts.ascending("_id"));

有关部分提及的方法和类的更多信息，请参阅以下资源：

UpdateOneModel API documentation
UpdateManyModel API 文档
UpdateOptions API documentation
唯一索引服务器手册说明

删除操作

要执行删除操作，请创建 DeleteOneModel 或 DeleteManyModel，指定要删除文档的查询筛选器。

DeleteOneModel 删除与查询筛选器匹配的第一个文档，DeleteManyModel 删除与查询筛选器匹配的所有文档。

重要

执行 bulkWrite() 时，如果没有与查询筛选器匹配的文件，则 DeleteOneModel 和 DeleteManyModel 不会删除任何文件。

例子

以下示例创建 DeleteOneModel 来删除 _id 为 1 的文档：

DeleteOneModel<Document> deleteDoc = new DeleteOneModel<>(Filters.eq("_id", 1));

有关本节中提到的方法和类的详情，请参阅以下 API 文档：

执行顺序

bulkWrite() 方法接受可选的 BulkWriteOptions 作为第二个参数，以指定批量操作是有序执行还是无序执行。

命令执行

默认情况下，bulkWrite() 方法按顺序执行批量操作。这意味着批量操作将按照您添加到列表中的顺序执行，直到出现错误（如有）。

例子

以下示例执行这些批量操作：

一个操作，插入 name 值为 "Zaynab Omar" 且 age 值为 37 的文档
一个操作，将 _id 为 1 的文档替换为包含 location 字段的新文档
一个操作，使用 "Zaynab Omar" 的 name 值更新文档并将 name 变更为 "Zaynab Hassan"
一个操作，删除所有 age 值大于 50 的文档

List<WriteModel<Document>> bulkOperations = new ArrayList<>();
// Creates instructions to insert a document
InsertOneModel<Document> insertDoc = new InsertOneModel<>(new Document("_id", 6)
                                                        .append("name", "Zaynab Omar")
                                                        .append("age", 37));
// Creates instructions to replace the first document matched by the query                                       
ReplaceOneModel<Document> replaceDoc = new ReplaceOneModel<>(Filters.eq("_id", 1), 
                                        new Document("name", "Sandy Kane")
                                            .append("location", "Helena, MT")); 
// Creates instructions to update the first document matched by the query                                                                 
UpdateOneModel<Document> updateDoc = new UpdateOneModel<>(Filters.eq("name", "Zaynab Omar"), 
                                        Updates.set("name", "Zaynab Hassan"));
// Creates instructions to delete all documents matched by the query                                
DeleteManyModel<Document> deleteDoc = new DeleteManyModel<>(Filters.gt("age", 50));
bulkOperations.add(insertDoc);
bulkOperations.add(replaceDoc);
bulkOperations.add(updateDoc);
bulkOperations.add(deleteDoc);
// Runs a bulk write operation for the specified the insert, replace, update, and delete WriteModels in order
collection.bulkWrite(bulkOperations);

运行此示例后，集合将包含以下文档：

{ "_id": 1, "name": "Sandy Kane", "location": "Helena, MT" }
{ "_id": 8, "name": "Shayla Ray", "age": 20 }
{ "_id": 6, "name": "Zaynab Hassan", "age": 37 }

无序执行

您还可以在 BulkWriteOptions 的 order() 方法中指定“false”，从而以任意顺序执行批量操作。这意味着无论是否出现错误，所有写操作都会执行，而在出现任何错误的情况下，将在最后报告批量操作。

在前面示例的基础上，添加以下内容，指定以任意顺序执行批量操作：

BulkWriteOptions options = new BulkWriteOptions().ordered(false);
// Runs a bulk write operation for the specified insert, replace, update, and delete WriteModels in any order
collection.bulkWrite(bulkOperations, options);

注意

无序批量操作不保证执行顺序。为了优化运行时间，顺序可以与您列出的方式不同。

在前一示例中，如果 bulkWrite() 方法决定在更新操作之后执行插入操作，则更新操作不会导致任何变化，因为该文档当时并不存在。随后，您的集合会包含以下文档：

{ "_id": 1, "name": "Sandy Kane", "location": "Helena, MT" }
{ "_id": 8, "name": "Shayla Ray", "age": 20 }
{ "_id": 6, "name": "Zaynab Omar", "age": 37 }

有关本节中提到的方法和类的详情，请参阅以下 API 文档：

批量写入示例：完整文件

注意

设置示例

This example connects to an instance of MongoDB by using a connection URI. To learn more about connecting to your MongoDB instance, see the Create a MongoClient guide. This example also uses the movies collection in the sample_mflix database included in the Atlas sample datasets. You can load them into your database on the free tier of MongoDB Atlas by following the MongoDB Get Started.

以下代码是完整的独立运行文件，用于执行以下操作：

创建 InsertOneModel、UpdateOneModel、DeleteOneModel 和 ReplaceOneModel 类的实例列表。
运行有序的 bulkWrite() 操作，以执行模型列表中指定的写入。

// Runs bulk write operations on a collection by using the Java driver
package org.example;
import java.util.Arrays;
import org.bson.Document;
import com.mongodb.MongoException;
import com.mongodb.bulk.BulkWriteResult;
import com.mongodb.client.MongoClient;
import com.mongodb.client.MongoClients;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoDatabase;
import com.mongodb.client.model.DeleteOneModel;
import com.mongodb.client.model.InsertOneModel;
import com.mongodb.client.model.ReplaceOneModel;
import com.mongodb.client.model.UpdateOneModel;
import com.mongodb.client.model.UpdateOptions;
public class BulkWrite {
    public static void main(String[] args) {
        // Replace the uri string with your MongoDB deployment's connection string
        String uri = "<connection string uri>";
        try (MongoClient mongoClient = MongoClients.create(uri)) {
            MongoDatabase database = mongoClient.getDatabase("sample_mflix");
            MongoCollection<Document> collection = database.getCollection("movies");
            // Runs a bulk write operation for the specified insert, update, delete, and replace operations
            BulkWriteResult result = collection.bulkWrite(
                    Arrays.asList(
                            new InsertOneModel<>(new Document("name", "A Sample Movie")),
                            new InsertOneModel<>(new Document("name", "Another Sample Movie")),
                            new InsertOneModel<>(new Document("name", "Yet Another Sample Movie")),
                            new UpdateOneModel<>(new Document("name", "A Sample Movie"),
                                    new Document("$set", new Document("name", "An Old Sample Movie")),
                                    new UpdateOptions().upsert(true)),
                            new DeleteOneModel<>(new Document("name", "Yet Another Sample Movie")),
                            new ReplaceOneModel<>(new Document("name", "Yet Another Sample Movie"),
                                    new Document("name", "The Other Sample Movie").append("runtime",  "42"))
                            ));
            // Prints the number of inserted, updated, and deleted documents
            System.out.println("Result statistics:" +
                    "\ninserted: " + result.getInsertedCount() +
                    "\nupdated: " + result.getModifiedCount() +
                    "\ndeleted: " + result.getDeletedCount());
        }
    }
}

Result statistics:
inserted: 3
updated: 2
deleted: 1

客户端批量写入

连接到运行MongoDB Server 8.0 或更高版本的部署时，可以使用 MongoClient.bulkWrite() 方法写入同一集群中的多个数据库和集合。 MongoClient.bulkWrite() 方法在一次调用中执行所有写入。

MongoClient.bulkWrite() 方法采用 ClientNamespacedWriteModel 实例列表来表示不同的写入操作。您可以使用实例方法构造 ClientNamespacedWriteModel 接口的实例。示例，ClientNamespacedInsertOneModel 的实例表示插入一个文档的操作，您可以使用 ClientNamespacedWriteModel.insertOne() 方法创建此模型。

注意

批量写入错误

如果任何写入操作失败，驱动程序都会引发 ClientBulkWriteException，并且不会执行任何进一步的单个操作。ClientBulkWriteException 包括可使用 ClientBulkWriteException.getWriteErrors() 方法访问的 BulkWriteError，其中提供了单个故障的详细信息。

下表描述了模型及其相应的实例方法。

模型	实例方法	说明	参数
`ClientNamespacedInsertOneModel`	`insertOne()`	创建一个模型以将文档插入到 `namespace` 中。	`namespace`：要写入的数据库和集合 `document`：要插入的文档
`ClientNamespacedUpdateOneModel`	`updateOne()`	创建一个模型以更新`namespace` 中与 `filter` 匹配的第一个文档。	`namespace`：要写入的数据库和集合 `filter`：用于选择要更新的文档的筛选器 `update`：更新以应用匹配文档 `updatePipeline`：更新管道以应用匹配文档 `options`：（可选）更新文档时应用的选项您必须为 `update` 或 `updatePipeline` 参数传递一个值。
`ClientNamespacedUpdateManyModel`	`updateMany()`	创建一个模型以更新`namespace` 中与 `filter` 匹配的所有文档。	`namespace`：要写入的数据库和集合 `filter`：用于选择要更新的文档的筛选器 `update`：更新以应用匹配文档 `updatePipeline`：更新管道以应用匹配文档 `options`：（可选）更新文档时应用的选项您必须为 `update` 或 `updatePipeline` 参数传递一个值。
`ClientNamespacedReplaceOneModel`	`replaceOne()`	创建一个模型以替换 `namespace` 中与 `filter` 匹配的第一个文档。	`namespace`：要写入的数据库和集合 `filter`：用于选择要替换的文档的筛选器 `replacement`：替换文档 `options`：（可选）替换文档时应用的选项
`ClientNamespacedDeleteOneModel`	`deleteOne()`	创建模型以删除`namespace` 中与 `filter` 匹配的第一个文档。	`namespace`：要写入的数据库和集合 `filter`：用于选择要删除的文档的筛选器 `option`：（可选）删除文档时应用的选项
`ClientNamespacedDeleteManyModel`	`deleteMany()`	创建模型以删除`namespace` 中与 `filter` 匹配的所有文档。	`namespace`：要写入的数据库和集合 `filter`：筛选器，用于选择要删除的文档 `option`：（可选）删除文档时应用的选项

以下部分举例说明如何使用客户端bulkWrite() 方法。

要学习；了解有关本节中提到的方法和类的更多信息，请参阅以下API文档：

插入示例

此示例演示如何使用 bulkWrite() 方法插入两个文档。将一个文档插入到 db.people集合中，而将另一文档插入到 db.things集合中。 MongoNamespace实例定义每个写入操作适用的数据库和集合。

MongoNamespace peopleNamespace = new MongoNamespace("db", "people");
MongoNamespace thingsNamespace = new MongoNamespace("db", "things");
List<ClientNamespacedWriteModel> bulkOperations = new ArrayList<>();
bulkOperations.add(ClientNamespacedWriteModel
        .insertOne(
                peopleNamespace,
                new Document("name", "Julia Smith")
        )
);
bulkOperations.add(ClientNamespacedWriteModel
        .insertOne(
                thingsNamespace,
                new Document("object", "washing machine")
        )
);
ClientBulkWriteResult result = mongoClient.bulkWrite(bulkOperations);

更新示例

以下示例展示如何使用 bulkWrite() 方法更新db.people 和 db.things 集合中的现有文档：

MongoNamespace peopleNamespace = new MongoNamespace("db", "people");
MongoNamespace thingsNamespace = new MongoNamespace("db", "things");
List<ClientNamespacedWriteModel> bulkOperations = new ArrayList<>();
bulkOperations.add(ClientNamespacedWriteModel.updateOne(
                peopleNamespace,
                Filters.eq("name", "Freya Polk"),
                Updates.inc("age", 1)
        )
);
bulkOperations.add(ClientNamespacedWriteModel.updateMany(
                thingsNamespace,
                Filters.eq("category", "electronic"),
                Updates.set("manufacturer", "Premium Technologies")
        )
);
ClientBulkWriteResult result = mongoClient.bulkWrite(bulkOperations);

此示例将 people集合中 name 值为 "Freya Polk" 的文档中 age字段的值递增 1。它还将 things集合中 category 值为 "electronic" 的所有文档的 manufacturer字段的值设置为 "Premium Technologies"。

如果多个文档与 ClientNamespacedUpdateOneModel实例中指定的查询筛选条件匹配，则该操作会更新第一个结果。您可以在 ClientUpdateOneOptions 实例中指定排序顺序，以便在驱动程序执行更新操作之前对匹配的文档应用顺序，如以下代码所示：

ClientUpdateOneOptions options = ClientUpdateOneOptions
        .clientUpdateOneOptions()
        .sort(Sorts.ascending("_id"));

替换示例

以下示例演示如何使用 bulkWrite() 方法替换 db.people 和 db.things 集合中的现有文档：

MongoNamespace peopleNamespace = new MongoNamespace("db", "people");
MongoNamespace thingsNamespace = new MongoNamespace("db", "things");
List<ClientNamespacedWriteModel> bulkOperations = new ArrayList<>();
bulkOperations.add(ClientNamespacedWriteModel.replaceOne(
                peopleNamespace,
                Filters.eq("_id", 1),
                new Document("name", "Frederic Hilbert")
        )
);
bulkOperations.add(ClientNamespacedWriteModel.replaceOne(
                thingsNamespace,
                Filters.eq("_id", 1),
                new Document("object", "potato")
        )
);
ClientBulkWriteResult result = mongoClient.bulkWrite(bulkOperations);

此示例成功运行后，people集合中 _id 值为 1 的文档将替换为新文档。 things集合中 _id 值为 1 的文档将替换为新文档。

如果多个文档与 ClientNamespacedReplaceOneModel实例中指定的查询筛选条件匹配，则该操作将替换第一个结果。您可以在 ClientReplaceOneOptions 实例中指定排序顺序，以便在驱动程序执行替换操作之前对匹配的文档应用顺序，如以下代码所示：

ClientReplaceOneOptions options = ClientReplaceOneOptions
        .clientReplaceOneOptions()
        .sort(Sorts.ascending("_id"));

BulkWriteOptions

您可以将 ClientBulkWriteOptions 的实例传递给 bulkWrite() 方法，以在运行批量写入操作时指定选项。

执行顺序示例

默认情况下，批量操作中的各个操作按照指定的顺序执行，直到出现错误或成功执行。不过，您可以将 false 传递给 ClientBulkWriteOptions 接口上的 ordered() 方法，以便以无序方式执行写入操作。使用无序选项时，产生错误的操作不会阻止在调用 bulkWrite() 方法时执行其他写入。

以下代码在 ClientBulkWriteOptions 的实例上设置 ordered() 方法，并执行批量写入操作以插入多个文档。

MongoNamespace namespace = new MongoNamespace("db", "people");
ClientBulkWriteOptions options = ClientBulkWriteOptions
        .clientBulkWriteOptions()
        .ordered(false);
List<ClientNamespacedWriteModel> bulkOperations = new ArrayList<>();
bulkOperations.add(
        ClientNamespacedWriteModel.insertOne(
                namespace,
                new Document("_id", 1).append("name", "Rudra Suraj")
        )
);
// Causes a duplicate key error
bulkOperations.add(
        ClientNamespacedWriteModel.insertOne(
                namespace,
                new Document("_id", 1).append("name", "Mario Bianchi")
        )
);
bulkOperations.add(
        ClientNamespacedWriteModel.insertOne(
                namespace,
                new Document("name", "Wendy Zhang")
        )
);
ClientBulkWriteResult result = mongoClient.bulkWrite(bulkOperations, options);

即使插入具有重复键的文档的写入操作会导致错误，也会执行其他操作，因为写入操作是无序的。

故障排除

BulkWriteException

如果驱动程序在批量写入操作期间遇到错误，驱动程序将抛出 MongoBulkWriteException。MongoBulkWriteException 包含一个 writeErrors 字段，该字段由与同一批量写入操作关联的一个或多个 WriteError 对象的列表组成。

考虑一个具有模式验证规则的集合，其中 quantity字段的值必须是 int 类型。在以下示例中，当您尝试插入一个 quantity字段值为 "three" 的文档和另一个 quantity字段值为 "ten" 的文档时，驱动程序会抛出一个 MongoBulkWriteException。

Exception in thread "main" com.mongodb.MongoBulkWriteException: Bulk write
operation result had errors at
com.mongodb.internal.connection.ProtocolHelper.getBulkWriteException(ProtocolHelper.java:258)
... at
BulkWriteMultipleValidationErrorsExample.main(BulkWriteMultipleValidationErrorsExample.java:30)
Caused by: com.mongodb.MongoWriteException: WriteError{code=121,
message='Document failed validation', details={ operator: "$jsonSchema",
schemaRules: { bsonType: "int", description: "must be an integer" },
offendingDocument: {"name":"Apple","quantity":"three"} }} at
com.mongodb.internal.connection.WriteResultHelper.createWriteException(WriteResultHelper.java:50)
at com.mongodb.internal.connection.ProtocolHelper.getBulkWriteException(ProtocolHelper.java:254)
... 19 more
Caused by: com.mongodb.MongoWriteException: WriteError{code=121,
message='Document failed validation', details={ operator: "$jsonSchema",
schemaRules: { bsonType: "int", description: "must be an integer" },
offendingDocument: {"name":"Banana","quantity":"ten"} }} at
com.mongodb.internal.connection.WriteResultHelper.createWriteException(WriteResultHelper.java:50)
at
com.mongodb.internal.connection.ProtocolHelper.getBulkWriteException(ProtocolHelper.java:254)
... 19 more

要学习；了解有关模式验证的更多信息，请参阅服务器手册条目部分中的模式验证。